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

convert TPL data flow to ActionBlock

Browse source
This commit is contained in:
Collin M. Barrett 2019-06-28 15:39:18 -05:00
parent bffef0c548
commit ba32b35aab
4 changed files with 47 additions and 77 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs
View file

@ -1,17 +0,0 @@
using System.Net.Http;
namespace FilterLists.Agent.Entities.Aggregates
{
public class ListDownload
{
public ListDownload(ListInfo listInfo, HttpResponseMessage httpResponseMessage)
{
ListInfo = listInfo;
HttpResponseMessage = httpResponseMessage;
}
public ListInfo ListInfo { get; }
public HttpResponseMessage HttpResponseMessage { get; }
}
}

49
src/FilterLists.Agent/ListArchiver/DownloadList.cs
View file

@ -4,29 +4,30 @@
using System.Threading;
using System.Threading.Tasks;
using FilterLists.Agent.Entities;
using FilterLists.Agent.Entities.Aggregates;
using FilterLists.Agent.ListArchiver.DownloadRequestsByFileExtension;
using MediatR;
using Microsoft.Extensions.Logging;
//TODO: upsert into MariaDB Rules table https://stackoverflow.com/questions/15271202/mysql-load-data-infile-with-on-duplicate-key-update
namespace FilterLists.Agent.ListArchiver
{
public static class DownloadList
{
public class Command : IRequest
{
public Command(ListDownload listDownload)
public Command(ListInfo listInfo)
{
ListDownload = listDownload;
ListInfo = listInfo;
}
public ListDownload ListDownload { get; }
public ListInfo ListInfo { get; }
}
public class Handler : AsyncRequestHandler<Command>
{
private static readonly Dictionary<string, Func<ListDownload, IRequest>> DownloadRequestsByFileExtension
= new Dictionary<string, Func<ListDownload, IRequest>>
private static readonly Dictionary<string, Func<ListInfo, IRequest>> DownloadRequestsByFileExtension
= new Dictionary<string, Func<ListInfo, IRequest>>
{
{"", l => new DownloadTxt.Command(l)},
{".7z", l => throw new NotImplementedException()},
@ -68,31 +69,29 @@ protected override async Task Handle(Command request, CancellationToken cancella
{
try
{
var extension = Path.GetExtension(request.ListDownload.ListInfo.ViewUrl.AbsolutePath);
if (DownloadRequestsByFileExtension.ContainsKey(extension))
{
_logger.LogInformation(
$"Downloading list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}...");
await _mediator.Send(DownloadRequestsByFileExtension[extension].Invoke(request.ListDownload),
cancellationToken);
}
else
{
_logger.LogWarning(
$"File extension not recognized for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
}
//TODO: upsert into MariaDB Rules table https://stackoverflow.com/questions/15271202/mysql-load-data-infile-with-on-duplicate-key-update
await DownloadByFileExtension(request, cancellationToken);
}
catch (NotImplementedException)
{
_logger.LogWarning(
$"File extension not supported for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
_logger.LogWarning($"File extension not supported for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
}
catch (ArgumentException ex)
{
_logger.LogError(ex,
$"Could not determine the file extension for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
_logger.LogError(ex, $"Could not determine the file extension for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
}
}
private async Task DownloadByFileExtension(Command request, CancellationToken cancellationToken)
{
var extension = Path.GetExtension(request.ListInfo.ViewUrl.AbsolutePath);
if (DownloadRequestsByFileExtension.ContainsKey(extension))
{
_logger.LogInformation($"Downloading list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}...");
await _mediator.Send(DownloadRequestsByFileExtension[extension].Invoke(request.ListInfo), cancellationToken);
}
else
{
_logger.LogWarning($"File extension not recognized for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
}
}
}

23
src/FilterLists.Agent/ListArchiver/DownloadLists.cs
View file

@ -1,10 +1,9 @@
using System.Collections.Generic;
using System.Net.Http;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using FilterLists.Agent.Entities;
using FilterLists.Agent.Entities.Aggregates;
using MediatR;
namespace FilterLists.Agent.ListArchiver
@ -23,36 +22,24 @@ public Command(IEnumerable<ListInfo> listInfo)
public class Handler : AsyncRequestHandler<Command>
{
private const int MaxDegreeOfParallelism = 5;
private readonly HttpClient _httpClient;
private const int MaxDegreeOfParallelism = 25;
private readonly IMediator _mediator;
public Handler(HttpClient httpClient, IMediator mediator)
public Handler(IMediator mediator)
{
_httpClient = httpClient;
_mediator = mediator;
}
// https://stackoverflow.com/a/22492731/2343739
protected override async Task Handle(Command request, CancellationToken cancellationToken)
{
var downloader = new TransformBlock<ListInfo, ListDownload>(
async l => new ListDownload(l, await _httpClient.GetAsync(l.ViewUrl, cancellationToken)),
var downloader = new ActionBlock<ListInfo>(
async l => await _mediator.Send(new DownloadList.Command(l), cancellationToken),
new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism}
);
var buffer = new BufferBlock<ListDownload>();
downloader.LinkTo(buffer);
foreach (var list in request.ListInfo)
await downloader.SendAsync(list, cancellationToken);
downloader.Complete();
foreach (var _ in request.ListInfo)
{
var listDownload = await buffer.ReceiveAsync(cancellationToken);
await _mediator.Send(new DownloadList.Command(listDownload), cancellationToken);
listDownload.HttpResponseMessage.Dispose();
}
await downloader.Completion;
}
}

35
src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs
View file

@ -2,7 +2,7 @@
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using FilterLists.Agent.Entities.Aggregates;
using FilterLists.Agent.Entities;
using MediatR;
using Microsoft.Extensions.Logging;
@ -12,12 +12,12 @@ public static class DownloadTxt
{
public class Command : IRequest
{
public Command(ListDownload listDownload)
public Command(ListInfo listInfo)
{
ListDownload = listDownload;
ListInfo = listInfo;
}
public ListDownload ListDownload { get; }
public ListInfo ListInfo { get; }
}
public class Handler : AsyncRequestHandler<Command>
@ -25,28 +25,29 @@ public class Handler : AsyncRequestHandler<Command>
private readonly HttpClient _httpClient;
private readonly ILogger<Handler> _logger;
public Handler(ILogger<Handler> logger, HttpClient httpClient)
public Handler(HttpClient httpClient, ILogger<Handler> logger)
{
_logger = logger;
_httpClient = httpClient;
_logger = logger;
}
protected override async Task Handle(Command request, CancellationToken cancellationToken)
{
try
using (var response = await _httpClient.GetAsync(request.ListInfo.ViewUrl, cancellationToken))
{
if (request.ListDownload.HttpResponseMessage.IsSuccessStatusCode)
using (Stream output =
File.OpenWrite(Path.Combine("archives", $"{request.ListDownload.ListInfo.Id}.txt")))
using (var input = await request.ListDownload.HttpResponseMessage.Content.ReadAsStreamAsync())
if (response.IsSuccessStatusCode)
try
{
input.CopyTo(output);
using (Stream output = File.OpenWrite(Path.Combine("archives", $"{request.ListInfo.Id}.txt")))
using (var input = await response.Content.ReadAsStreamAsync())
{
input.CopyTo(output);
}
}
catch (HttpRequestException ex)
{
_logger.LogError(ex, $"Error downloading list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
}
}
catch (HttpRequestException ex)
{
_logger.LogError(ex,
$"Error downloading list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
}
}
}