Задача SendAsync потока данных TPL никогда не завершается, когда блок связан

Я столкнулся с той же проблемой, что и вы. Я не углублялся в реализацию LinkTo, но я думаю, что сообщение распространяется только тогда, когда исходный блок его получил. Я имею в виду, что может быть случай, когда исходный блок имеет несколько сообщений на входе, но он не будет их обрабатывать до следующего полученного сообщения Post / SendAsync. И это ваше дело.

Вот мое решение, и оно работает для меня.

Сначала объявите "двигатель"

/// <summary>
/// Engine-class (like a car engine) that produced a lot count (or infinite) of actions.
/// </summary>
public class Engine
{
    private BufferBlock<int> _bufferBlock;

    /// <summary>
    /// Creates source block that produced stub data.
    /// </summary>
    /// <param name="count">Count of actions. If count = 0 then it's infinite loop.</param>
    /// <param name="boundedCapacity">Bounded capacity (throttling).</param>
    /// <param name="cancellationToken">Cancellation token (used to stop infinite loop).</param>
    /// <returns>Source block that constantly produced 0-value.</returns>
    public ISourceBlock<int> CreateEngine(int count, int boundedCapacity, CancellationToken cancellationToken)
    {
        _bufferBlock = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = boundedCapacity });

        Task.Run(async () =>
        {
            var counter = 0;
            while (count == 0 || counter < count)
            {
                await _bufferBlock.SendAsync(0);
                if (cancellationToken.IsCancellationRequested)
                    return;
                counter++;
            }
        }, cancellationToken).ContinueWith((task) =>
        {
            _bufferBlock.Complete();
        });

        return _bufferBlock;
    }
}

Затем продюсер, использующий движок

/// <summary>
/// Producer that generates random byte blobs with specified size.
/// </summary>
public class Producer
{
    private static Random random = new Random();

    /// <summary>
    /// Returns source block that produced byte arrays. 
    /// </summary>
    /// <param name="blobSize">Size of byte arrays.</param>
    /// <param name="count">Total count of blobs (if 0 then infinite).</param>
    /// <param name="boundedCapacity">Bounded capacity (throttling).</param>
    /// <param name="cancellationToken">Cancellation token (used to stop infinite loop).</param>
    /// <returns>Source block.</returns>
    public static ISourceBlock<byte[]> BlobsSourceBlock(int blobSize, int count, int boundedCapacity, CancellationToken cancellationToken)
    {
        // Creating engine with specified bounded capacity.
        var engine = new Engine().CreateEngine(count, boundedCapacity, cancellationToken);

        // Creating transform block that uses our driver as a source.
        var block = new TransformBlock<int, byte[]>(
            // Useful work.
            i => CreateBlob(blobSize),
            new ExecutionDataflowBlockOptions
            {
                // Here you can specify your own throttling. 
                BoundedCapacity = boundedCapacity,
                MaxDegreeOfParallelism = Environment.ProcessorCount,
            });
        // Linking engine (and engine is already working at that time).
        engine.LinkTo(block, new DataflowLinkOptions { PropagateCompletion = true });
        return block;
    }

    /// <summary>
    /// Simple random byte[] generator.
    /// </summary>
    /// <param name="size">Array size.</param>
    /// <returns>byte[]</returns>
    private static byte[] CreateBlob(int size)
    {
        var buffer = new byte[size];
        random.NextBytes(buffer);
        return buffer;
    }
}

Теперь вы можете использовать производителя с потребителем (например, ActionBlock)

        var blobsProducer = BlobsProducer.CreateAndStartBlobsSourceBlock(0, 1024 * 1024, 10, cancellationTokenSource.Token);

        var md5Hash = MD5.Create();

        var actionBlock = new ActionBlock<byte[]>(b => 
        {
            Console.WriteLine(GetMd5Hash(md5Hash, b));
        },
        new ExecutionDataflowBlockOptions() { BoundedCapacity = 10 });

        blobsProducer.LinkTo(actionBlock);

Надеюсь, это поможет вам!