From ba32b35aab9636aaabd6366961b770726ff1a11d Mon Sep 17 00:00:00 2001 From: "Collin M. Barrett" Date: Fri, 28 Jun 2019 15:39:18 -0500 Subject: [PATCH] convert TPL data flow to ActionBlock --- .../Entities/Aggregates/ListDownload.cs | 17 ------- .../ListArchiver/DownloadList.cs | 49 +++++++++---------- .../ListArchiver/DownloadLists.cs | 23 ++------- .../DownloadTxt.cs | 35 ++++++------- 4 files changed, 47 insertions(+), 77 deletions(-) delete mode 100644 src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs diff --git a/src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs b/src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs deleted file mode 100644 index 717b31b7b..000000000 --- a/src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs +++ /dev/null @@ -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; } - } -} \ No newline at end of file diff --git a/src/FilterLists.Agent/ListArchiver/DownloadList.cs b/src/FilterLists.Agent/ListArchiver/DownloadList.cs index a8b73cd79..7f4815778 100644 --- a/src/FilterLists.Agent/ListArchiver/DownloadList.cs +++ b/src/FilterLists.Agent/ListArchiver/DownloadList.cs @@ -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 { - private static readonly Dictionary> DownloadRequestsByFileExtension - = new Dictionary> + private static readonly Dictionary> DownloadRequestsByFileExtension + = new Dictionary> { {"", 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}."); } } } diff --git a/src/FilterLists.Agent/ListArchiver/DownloadLists.cs b/src/FilterLists.Agent/ListArchiver/DownloadLists.cs index d1f5008b9..dbf9f3f29 100644 --- a/src/FilterLists.Agent/ListArchiver/DownloadLists.cs +++ b/src/FilterLists.Agent/ListArchiver/DownloadLists.cs @@ -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) public class Handler : AsyncRequestHandler { - 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( - async l => new ListDownload(l, await _httpClient.GetAsync(l.ViewUrl, cancellationToken)), + var downloader = new ActionBlock( + async l => await _mediator.Send(new DownloadList.Command(l), cancellationToken), new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism} ); - var buffer = new BufferBlock(); - 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; } } diff --git a/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs b/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs index 819ebece1..89f05f1d2 100644 --- a/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs +++ b/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs @@ -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 @@ -25,28 +25,29 @@ public class Handler : AsyncRequestHandler private readonly HttpClient _httpClient; private readonly ILogger _logger; - public Handler(ILogger logger, HttpClient httpClient) + public Handler(HttpClient httpClient, ILogger 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}."); } } }