using System;

using System.Linq;

using System.Reactive;

using System.Reactive.Linq;

using System.Collections.Generic;

using System.Threading.Tasks;

using System.Threading.Tasks.Dataflow;

public class Program

{

    public static async Task Main()

    {

        Pipeline p = new();

        using var subscription = Observable.Range(0, 55).Subscribe(p.InputObserver);

        await p.Completion();

    }

}

​

public class Pipeline

{

    const int BATCHSIZE = 10;

    private readonly TransformBlock<int, Item> _idToItemBlock;

    private readonly ActionBlock<IEnumerable<Item>> _itemProcessor;

    public Pipeline()

    {

        var executionBlockOptions = new ExecutionDataflowBlockOptions {

                    BoundedCapacity = BATCHSIZE

                };

        _idToItemBlock = new TransformBlock<int, Item>(IdToItemAsync, executionBlockOptions);

        var itemBatcher = new BatchBlock<Item>(BATCHSIZE);

        _itemProcessor = new ActionBlock<IEnumerable<Item>>(ProcessEmailsAsync, executionBlockOptions);

        _idToItemBlock.LinkTo(itemBatcher, new DataflowLinkOptions { PropagateCompletion = true });

        itemBatcher.LinkTo(_itemProcessor, new DataflowLinkOptions { PropagateCompletion = true });

    }

    private static async Task<Item> IdToItemAsync(int id) 

    {

        await Task.Yield();

        return new Item(id);

    }

    private static async Task ProcessEmailsAsync(IEnumerable<Item> items)

    {

        var itemsList = items.ToList();

        Console.WriteLine($"Processing Batch of Size {itemsList.Count}");

        foreach ( var item in itemsList )

        {

            await Task.Delay(25); // Do some work

            Console.WriteLine($"Processed {item}");

        }

    }

    public IObserver<int> InputObserver => _idToItemBlock.AsObserver();

    public Task Completion() => _itemProcessor.Completion;

}

​

public record Item( int Id );