Beam/Beam/UnitFragmentDownloader.cs

using HtmlAgilityPack;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Beam {
    public class UnitFragmentDownloader<T> : IUnitDownloader<Fragment<Ordered<T>>> {
        public UnitFragmentDownloader(HtmlWeb web,
                                      AsyncHtmlTransformer<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 AsyncHtmlTransformer<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(Ordered<string>[] link, CancellationToken ct, int maximumRetryCount, IProgress<RetryReport>? tryProgress) {
            Fragment<Ordered<T>> fragment = new Fragment<Ordered<T>>(link.Length);
            if (!Fragment<Ordered<T>>.TryAcquireUpdater(fragment, out var updater))
                throw new S.AssertionException(S.M.NewFragmentShouldBeFree);
            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 {} order={}", x.Data, x.Order);
                    return;
                }
                if (downloadedT == null) {
                    Interlocked.Exchange(ref isFailure, true);
                    Logger?.LogCritical("Failed to retrieve {} order={}", 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);

        }
    }
}