prehistoric-systems/FilterLists
1
0
Fork 0
mirror of https://github.com/collinbarrett/FilterLists.git synced 2026-03-11 09:04:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(archival): ✨⚡♻ support segments of different source types, use IAsyncEnumerable for Streams

Browse source
This commit is contained in:
Collin M. Barrett 2020-09-18 16:57:28 -05:00
parent c4d97dd8b4
commit e95892af1b
11 changed files with 106 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
services/Archival/FilterLists.Archival.Application/Commands/ArchiveList.cs
View file

@ -2,6 +2,7 @@
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using FilterLists.Archival.Application.Models;
@ -53,7 +54,7 @@ public async Task<Unit> Handle(Command request, CancellationToken cancellationTo
var segmentUrls = await GetSegmentUrlsAsync(request.ListId, cancellationToken);
if (segmentUrls.Count > 0)
{
var file = await GetFileAsync(request.ListId, segmentUrls, cancellationToken);
var file = GetFileToArchive(request.ListId, segmentUrls, cancellationToken);
await _archiver.ArchiveFileAsync(file, cancellationToken);
_archiver.Commit();
@ -83,30 +84,28 @@ private async Task<List<ListDetailsViewUrlVm>> GetSegmentUrlsAsync(
new List<ListDetailsViewUrlVm>();
}
private async Task<FileToArchive> GetFileAsync(
private IFileToArchive GetFileToArchive(
int listId,
IReadOnlyCollection<ListDetailsViewUrlVm> segmentUrls,
CancellationToken cancellationToken)
{
var contentsAsync = GetContentsAsync(segmentUrls, cancellationToken);
var sourceFileName = Uri.UnescapeDataString(segmentUrls.First().Url.Segments.Last());
var sourceExtension = Path.GetExtension(sourceFileName);
var target = new FileInfo($"{listId}.txt");
return new FileToArchive(sourceExtension, await contentsAsync, target);
}
// TODO: IAsyncEnumerable?
private async Task<IEnumerable<Stream>> GetContentsAsync(
IEnumerable<ListDetailsViewUrlVm> segmentUrls,
CancellationToken cancellationToken)
{
var downloads = new List<Task<Stream>>();
var segmentsAsync = GetSegmentsAsync(segmentUrls, cancellationToken);
var target = new FileInfo($"{listId}.txt");
return new FileToArchive(segmentsAsync, target);
}
private async IAsyncEnumerable<IFileToArchiveSegment> GetSegmentsAsync(
IEnumerable<ListDetailsViewUrlVm> segmentUrls,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
foreach (var segment in segmentUrls)
{
downloads.Add(_client.DownloadFileAsync(segment.Url, cancellationToken));
var sourceFileName = Uri.UnescapeDataString(segment.Url.Segments.Last());
var sourceExtension = Path.GetExtension(sourceFileName);
yield return new FileToArchiveSegment(
sourceExtension,
await _client.DownloadFileAsync(segment.Url, cancellationToken));
}
return await Task.WhenAll(downloads);
}
}
}

20
services/Archival/FilterLists.Archival.Application/Models/FileToArchive.cs
View file

@ -6,15 +6,25 @@ namespace FilterLists.Archival.Application.Models
{
internal class FileToArchive : IFileToArchive
{
public FileToArchive(string sourceExtension, IEnumerable<Stream> contents, FileInfo target)
public FileToArchive(IAsyncEnumerable<IFileToArchiveSegment> segments, FileInfo target)
{
SourceExtension = sourceExtension;
Contents = contents;
Segments = segments;
Target = target;
}
public string SourceExtension { get; }
public IEnumerable<Stream> Contents { get; }
public IAsyncEnumerable<IFileToArchiveSegment> Segments { get; }
public FileInfo Target { get; }
}
internal class FileToArchiveSegment : IFileToArchiveSegment
{
public FileToArchiveSegment(string sourceExtension, Stream contents)
{
SourceExtension = sourceExtension;
Contents = contents;
}
public string SourceExtension { get; }
public Stream Contents { get; }
}
}

23
services/Archival/FilterLists.Archival.Infrastructure/Persistence/FileWriteStrategies/FileStreamConversionStrategyFactory.cs Normal file
View file

@ -0,0 +1,23 @@
using System;
using System.Linq;
using System.Reflection;
namespace FilterLists.Archival.Infrastructure.Persistence.FileWriteStrategies
{
internal static class FileStreamConversionStrategyFactory
{
public static TFileStreamConversionStrategy? GetStrategy<TFileStreamConversionStrategy>(
this IFileToArchiveSegment segment)
where TFileStreamConversionStrategy : class, IFileStreamConversionStrategy
{
var strategyType = Assembly.GetExecutingAssembly()
.GetTypes()
.FirstOrDefault(t =>
typeof(TFileStreamConversionStrategy).IsAssignableFrom(t) &&
string.Equals(t.Name, segment.SourceExtension.TrimStart('.'), StringComparison.OrdinalIgnoreCase));
return strategyType is default(Type)
? default
: (TFileStreamConversionStrategy)Activator.CreateInstance(strategyType);
}
}
}

22
services/Archival/FilterLists.Archival.Infrastructure/Persistence/FileWriteStrategies/FileWriteStrategyFactory.cs
View file

@ -1,22 +0,0 @@
using System;
using System.Linq;
using System.Reflection;
namespace FilterLists.Archival.Infrastructure.Persistence.FileWriteStrategies
{
internal static class FileWriteStrategyFactory
{
public static TFileWriteStrategy? GetStrategy<TFileWriteStrategy>(this IFileToArchive file)
where TFileWriteStrategy : class, IFileWriteStrategy
{
var strategyType = Assembly.GetExecutingAssembly()
.GetTypes()
.FirstOrDefault(t =>
typeof(TFileWriteStrategy).IsAssignableFrom(t) &&
string.Equals(t.Name, file.SourceExtension.TrimStart('.'), StringComparison.OrdinalIgnoreCase));
return strategyType is default(Type)
? default
: (TFileWriteStrategy)Activator.CreateInstance(strategyType);
}
}
}

10
services/Archival/FilterLists.Archival.Infrastructure/Persistence/FileWriteStrategies/IFileStreamConversionStrategy.cs Normal file
View file

@ -0,0 +1,10 @@
using System.IO;
using System.Threading;
namespace FilterLists.Archival.Infrastructure.Persistence.FileWriteStrategies
{
internal interface IFileStreamConversionStrategy
{
Stream Convert(IFileToArchiveSegment segment, CancellationToken cancellationToken);
}
}

10
services/Archival/FilterLists.Archival.Infrastructure/Persistence/FileWriteStrategies/IFileWriteStrategy.cs
View file

@ -1,10 +0,0 @@
using System.Threading;
using System.Threading.Tasks;
namespace FilterLists.Archival.Infrastructure.Persistence.FileWriteStrategies
{
internal interface IFileWriteStrategy
{
Task WriteAsync(IFileToArchive file, CancellationToken cancellationToken = default);
}
}

14
services/Archival/FilterLists.Archival.Infrastructure/Persistence/FileWriteStrategies/Txt.cs
View file

@ -1,20 +1,16 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace FilterLists.Archival.Infrastructure.Persistence.FileWriteStrategies
{
public class Txt : IFileWriteStrategy
public class Txt : IFileStreamConversionStrategy
{
public async Task WriteAsync(IFileToArchive file, CancellationToken cancellationToken = default)
public Stream Convert(IFileToArchiveSegment segment, CancellationToken cancellationToken)
{
_ = file ?? throw new ArgumentNullException(nameof(file));
_ = segment ?? throw new ArgumentNullException(nameof(segment));
await using var target = file.Target.OpenWrite();
foreach (var source in file.Contents)
{
await source.CopyToAsync(target, cancellationToken);
}
return segment.Contents;
}
}
}

45
services/Archival/FilterLists.Archival.Infrastructure/Persistence/GitFileArchiver.cs
View file

@ -31,29 +31,36 @@ public GitFileArchiver(
public async Task ArchiveFileAsync(
IFileToArchive file,
CancellationToken cancellationToken = default)
CancellationToken cancellationToken)
{
var strategy = file.GetStrategy<IFileWriteStrategy>();
if (strategy is default(IFileWriteStrategy))
var textStreams = new List<Stream>();
await foreach (var segment in file.Segments.WithCancellation(cancellationToken))
{
_logger.LogWarning(
"No write strategy found for {FileName}. Skipping",
file.Target.Name);
var strategy = segment.GetStrategy<IFileStreamConversionStrategy>();
if (strategy is default(IFileStreamConversionStrategy))
{
_logger.LogWarning(
"No file stream conversion strategy found for extension {Extension} in target {Target}. Skipping file",
segment.SourceExtension,
file.Target);
break;
}
textStreams.Add(strategy.Convert(segment, cancellationToken));
}
else
_logger.LogDebug("Writing {FileName}", file.Target.Name);
_writtenFiles.Add(file.Target);
// TODO: write to _options.RepositoryPath
await using var target = file.Target.OpenWrite();
foreach (var textStream in textStreams)
{
_logger.LogDebug(
"Writing {FileName} with strategy {FileWriteStrategy}",
file.Target.Name,
strategy.GetType().Name);
_writtenFiles.Add(file.Target);
// TODO: write to _options.RepositoryPath
await strategy.WriteAsync(file, cancellationToken);
_logger.LogDebug("Finished writing {FileName}", file.Target.Name);
await textStream.CopyToAsync(target, cancellationToken);
}
_logger.LogDebug("Finished writing {FileName}", file.Target.Name);
}
public void Commit()
@ -61,7 +68,7 @@ public void Commit()
var fileNames = _writtenFiles.Select(f => f.Name).ToList();
Commands.Stage(_repo, fileNames);
var signature = new Signature(_options.UserName, _options.UserEmail, DateTime.UtcNow);
var message = $"feat(archives): archive {fileNames.Count} files{Environment.NewLine}{string.Join(Environment.NewLine, fileNames)}";
var message = $"feat(archives): archive {fileNames.Count} files{Environment.NewLine}{string.Join(Environment.NewLine, fileNames)}";
_repo.Commit(message, signature, signature);
_logger.LogDebug("Committed {@FileNames}", fileNames);

2
services/Archival/FilterLists.Archival.Infrastructure/Persistence/IFileArchiver.cs
View file

@ -8,6 +8,6 @@ public interface IFileArchiver : IUnitOfWork
{
Task ArchiveFileAsync(
IFileToArchive file,
CancellationToken cancellationToken = default);
CancellationToken cancellationToken);
}
}

10
services/Archival/FilterLists.Archival.Infrastructure/Persistence/IFileToArchive.cs
View file

@ -5,9 +5,13 @@ namespace FilterLists.Archival.Infrastructure.Persistence
{
public interface IFileToArchive
{
// for now, assume all source segments have same extension
string SourceExtension { get; }
IEnumerable<Stream> Contents { get; }
IAsyncEnumerable<IFileToArchiveSegment> Segments { get; }
FileInfo Target { get; }
}
public interface IFileToArchiveSegment
{
string SourceExtension { get; }
Stream Contents { get; }
}
}

4
services/SharedKernel/FilterLists.SharedKernel.Apis.Clients/IDirectoryApi.cs
View file

@ -9,9 +9,9 @@ namespace FilterLists.SharedKernel.Apis.Clients
public interface IDirectoryApi
{
[Get("/lists")]
Task<IEnumerable<ListVm>> GetListsAsync(CancellationToken cancellationToken = default);
Task<IEnumerable<ListVm>> GetListsAsync(CancellationToken cancellationToken);
[Get("/lists/{id}")]
Task<ListDetailsVm> GetListDetailsAsync(int id, CancellationToken cancellationToken = default);
Task<ListDetailsVm> GetListDetailsAsync(int id, CancellationToken cancellationToken);
}
}