Beam/Beam.Downloaders/UnitFragmentDownloader.cs

using Beam.Abstractions;
using Beam.Exceptions;
using Beam.Models;
using HtmlAgilityPack;
using Microsoft.Extensions.Logging;

namespace Beam.Downloaders {
    public class UnitFragmentDownloader<T> : IUnitDownloader<Fragment<Ordered<T>>> {
        public UnitFragmentDownloader(HtmlWeb web,
                                      AsyncTransformer<HtmlDocument, T> transformer,
                                      AsyncDownloadFailurePredicate<HtmlDocument>?[]? failurePredicate = null,
                                      int fragmentSize = 4,
                                      ILogger? logger = null,
                                      IUnitDownloader<T>? internalDownloader = null) {
            Web = web;
            Transformer = transformer;
            FailurePredicate = failurePredicate;
            UnitDownloader = internalDownloader ?? new UnitDownloader<T>(Web, Transformer, FailurePredicate);
            LinksPerDownload = fragmentSize;
            Logger = logger;
        }

        public HtmlWeb Web { get; }
        public AsyncTransformer<HtmlDocument, T> Transformer { get; }
        public AsyncDownloadFailurePredicate<HtmlDocument>?[]? FailurePredicate { get; }
        public int LinksPerDownload { get; set; }
        public ILogger? Logger { get; set;  }

        private readonly IUnitDownloader<T> UnitDownloader;

        async Task<(bool, Fragment<Ordered<T>>?)> IUnitDownloader<Fragment<Ordered<T>>>.TryDownload(IOrdered<string>[] link, CancellationToken ct, int maximumRetryCount, IProgress<IRetryReport>? tryProgress) {
            Fragment<Ordered<T>> fragment = new Fragment<Ordered<T>>(link.Length);
            if (!Fragment<Ordered<T>>.TryAcquireUpdater(fragment, out var updater))
                throw new AssertionException(Exceptions.Exceptions.fragment_locked);
            bool isFailure = false;
            await Parallel.ForEachAsync(link, async (x, pct) => {
                pct.ThrowIfCancellationRequested();
                ct.ThrowIfCancellationRequested();
                var (result, downloadedT) = await UnitDownloader.TryDownload([x], ct, maximumRetryCount, tryProgress);
                if (!result) {
                    Interlocked.Exchange(ref isFailure, true);
                    Logger?.LogError("Failed to retrieve {0} order={1}", x.Data, x.Order);
                    return;
                }
                if (downloadedT == null) {
                    Interlocked.Exchange(ref isFailure, true);
                    Logger?.LogCritical("Failed to retrieve {0} order={1}", x.Data, x.Order);
                    return;
                }
                updater(new Ordered<T>(downloadedT, x.Order));
            });

            if (!isFailure)
                Fragment<Ordered<T>>.SetComplete(fragment, true);

            Fragment<Ordered<T>>.TryReleaseUpdater(fragment, updater);
            
            return (!isFailure, fragment);

        }
    }
}