MixedConsumerProducer Class

Namespace: Meziantou.Framework.Threading
Assembly: Meziantou.Framework.Threading.dll
Package: Meziantou.Framework.Threading 3.1.9
Source: MixedConsumerProducer.cs
Available in package assets: net10.0net11.0

Provides a parallel processing utility that allows consumers to dynamically add new items to be processed.

public static class MixedConsumerProducer

Methods

Name	Description
Process<T>(IEnumerable<T>, ParallelOptions, Func<MixedConsumerProducerContext<T>, T, CancellationToken, ValueTask>)	Processes items in parallel, where each processing action can enqueue additional items to be processed.

Process<T>(IEnumerable<T>, ParallelOptions, Func<MixedConsumerProducerContext<T>, T, CancellationToken, ValueTask>)

MixedConsumerProducer.cs

Processes items in parallel, where each processing action can enqueue additional items to be processed.

net10.0

public static Task Process<T>(IEnumerable<T> initialItems, ParallelOptions options, Func<MixedConsumerProducerContext<T>, T, CancellationToken, ValueTask> action)

net11.0

public static Task Process<T>(IEnumerable<T> initialItems, ParallelOptions options, Func<MixedConsumerProducerContext<T>, T, CancellationToken, ValueTask> action)

Type parameters

T: The type of items to process.

Parameters

initialItems IEnumerable<T>: The initial collection of items to process.
options ParallelOptions: Options that configure the parallel processing. ParallelOptions.MaxDegreeOfParallelism, ParallelOptions.CancellationToken, and ParallelOptions.TaskScheduler are honored. Pass a MixedConsumerProducerOptions to also limit the number of pending items with MixedConsumerProducerOptions.MaxPendingItems.
action Func<MixedConsumerProducerContext<T>, T, CancellationToken, ValueTask>: The action to perform on each item. The action receives a context to enqueue new items, the current item, and a cancellation token.

Returns

Task A task that represents the asynchronous operation.

Exceptions

AggregateException: One or more invocations of action threw an exception. Remaining items are still processed, and all exceptions are reported once processing completes.
OperationCanceledException: ParallelOptions.CancellationToken was canceled.
TaskSchedulerException: ParallelOptions.TaskScheduler rejected a consumer. The consumers that did start are completed before the exception is thrown.

Example

// A fake website where each page links to other pages
var website = new Dictionary<string, string[]>(StringComparer.Ordinal)
{
    ["https://example.com/"] = ["https://example.com/a", "https://example.com/b"],
    ["https://example.com/a"] = ["https://example.com/a/1"],
};
var crawledPages = new ConcurrentBag<string>();

async Task<string[]> CrawlPageAsync(string url, CancellationToken cancellationToken)
{
    await Task.Delay(1, cancellationToken);
    crawledPages.Add(url);
    return website.GetValueOrDefault(url, []);
}

var initialUrls = new[] { "https://example.com/" };
await MixedConsumerProducer.Process(
    initialUrls,
    new ParallelOptions { MaxDegreeOfParallelism = 4 },
    async (context, url, ct) =>
    {
        var links = await CrawlPageAsync(url, ct);
        foreach (var link in links)
        {
            context.Enqueue(link);
        }
    });

Assert.Equal(["https://example.com/", "https://example.com/a", "https://example.com/a/1", "https://example.com/b"], crawledPages.Order(StringComparer.Ordinal));