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

wip fix TPL Data Flow batch downloads

Browse source
This commit is contained in:
Collin M. Barrett 2019-06-27 22:13:45 -05:00
parent ed9ce8b7fe
commit 088f44272f
4 changed files with 60 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -0,0 +1,17 @@
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; }
}
}

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

@ -4,6 +4,7 @@
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;
@ -14,18 +15,18 @@ public static class DownloadList
{
public class Command : IRequest
{
public Command(ListInfo listInfo)
public Command(ListDownload listDownload)
{
ListInfo = listInfo;
ListDownload = listDownload;
}
public ListInfo ListInfo { get; }
public ListDownload ListDownload { get; }
}
public class Handler : AsyncRequestHandler<Command>
{
private static readonly Dictionary<string, Func<ListInfo, IRequest>> DownloadRequestsByFileExtension
= new Dictionary<string, Func<ListInfo, IRequest>>
private static readonly Dictionary<string, Func<ListDownload, IRequest>> DownloadRequestsByFileExtension
= new Dictionary<string, Func<ListDownload, IRequest>>
{
{"", l => new DownloadTxt.Command(l)},
{".7z", l => throw new NotImplementedException()},
@ -67,18 +68,18 @@ protected override async Task Handle(Command request, CancellationToken cancella
{
try
{
var extension = Path.GetExtension(request.ListInfo.ViewUrl.AbsolutePath);
var extension = Path.GetExtension(request.ListDownload.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),
$"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.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
$"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
@ -86,12 +87,12 @@ await _mediator.Send(DownloadRequestsByFileExtension[extension].Invoke(request.L
catch (NotImplementedException)
{
_logger.LogWarning(
$"File extension not supported for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
$"File extension not supported for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
}
catch (ArgumentException ex)
{
_logger.LogError(ex,
$"Could not determine the file extension for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
$"Could not determine the file extension for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
}
}
}

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

@ -1,8 +1,10 @@
using System.Collections.Generic;
using System.Net.Http;
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
@ -21,28 +23,37 @@ public Command(IEnumerable<ListInfo> lists)
public class Handler : AsyncRequestHandler<Command>
{
private const int MaxDegreeOfParallelism = 50;
private const int MaxDegreeOfParallelism = 5;
private readonly HttpClient _httpClient;
private readonly IMediator _mediator;
public Handler(IMediator mediator)
public Handler(IMediator mediator, HttpClient httpClient)
{
_mediator = mediator;
_httpClient = httpClient;
}
// https://stackoverflow.com/a/22492731/2343739
protected override async Task Handle(Command request, CancellationToken cancellationToken)
{
var downloader = new TransformBlock<ListInfo, Task>(
l => _mediator.Send(new DownloadList.Command(l), cancellationToken),
new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism});
var buffer = new BufferBlock<Task>();
var downloader = new TransformBlock<ListInfo, ListDownload>(
async l => new ListDownload(l, await _httpClient.GetAsync(l.ViewUrl, cancellationToken)),
new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism}
);
var buffer = new BufferBlock<ListDownload>();
downloader.LinkTo(buffer);
foreach (var list in request.Lists)
await downloader.SendAsync(list, cancellationToken);
downloader.Complete();
foreach (var _ in request.Lists)
{
var listDownload = await buffer.ReceiveAsync(cancellationToken);
await _mediator.Send(new DownloadList.Command(listDownload), cancellationToken);
listDownload.HttpResponseMessage.Dispose();
}
await downloader.Completion;
if (buffer.TryReceiveAll(out var tasks))
await Task.WhenAll(tasks);
}
}
}

27
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;
using FilterLists.Agent.Entities.Aggregates;
using MediatR;
using Microsoft.Extensions.Logging;
@ -12,12 +12,12 @@ public static class DownloadTxt
{
public class Command : IRequest
{
public Command(ListInfo listInfo)
public Command(ListDownload listDownload)
{
ListInfo = listInfo;
ListDownload = listDownload;
}
public ListInfo ListInfo { get; }
public ListDownload ListDownload { get; }
}
public class Handler : AsyncRequestHandler<Command>
@ -35,21 +35,18 @@ protected override async Task Handle(Command request, CancellationToken cancella
{
try
{
using (var result = await _httpClient.GetAsync(request.ListInfo.ViewUrl, cancellationToken))
{
if (result.IsSuccessStatusCode)
using (Stream output =
File.OpenWrite(Path.Combine("archives", $"{request.ListInfo.Id}.txt")))
using (var input = await result.Content.ReadAsStreamAsync())
{
input.CopyTo(output);
}
}
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())
{
input.CopyTo(output);
}
}
catch (HttpRequestException ex)
{
_logger.LogError(ex,
$"Error downloading list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}.");
$"Error downloading list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}.");
}
}
}