Custom TPL Block: how to test Consume, Release, Reserve

Question

Suppose I want to create a custom ISourceBlock<T>. From the MSDN sliding window example, I could begin the construction of my custom block like so:

public sealed class SomeCustomBlock<T> : ISourceBlock<T>
{
    private readonly int _somePrivateMember = 0;
    private readonly ISourceBlock<T> _source = new BufferBlock<T>();

    public void Complete() => _source.Complete();
    public void Fault(Exception exception) => _source.Fault(exception);
    public Task Completion => _source.Completion;
    
    public T? ConsumeMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target, out bool messageConsumed)
    {
        return _source.ConsumeMessage(messageHeader, target, out messageConsumed);
    }

    public IDisposable LinkTo(ITargetBlock<T> target, DataflowLinkOptions linkOptions)
    {
        return _source.LinkTo(target, linkOptions);
    }

    public void ReleaseReservation(DataflowMessageHeader messageHeader, ITargetBlock<T> target)
    {
        _source.ReleaseReservation(messageHeader, target);
    }

    public bool ReserveMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target)
    {
        return _source.ReserveMessage(messageHeader, target);
    }
}

The trouble is, since the implementation of the ISourceBlock<T> interface is forwarded to my private _source field, linked blocks will not call SomeCustomBlock's implementation of ConsumeMessage, ReleaseReservation, and ReserveMessage methods directly: they'll call the _source's, aka BufferBlock<T>'s implementation. In fact, I have had a heck of a time trying to get the SomeCustomBlock's implementation of these methods to be called at all; any target that calls these methods will communicate with _source since that is the block they directly link with.

My questions:

How can I write unit tests to verify that I've been a good developer and have not broken SomeCustomBlock's implementation of the ConsumeMessage, ReleaseReservation, and ReserveMessage methods? I work a lot with NUnit, but I'm not opposed to another testing framework if it's required.
Is there a better way to create a custom TPL block that I can more easily test fully, assuming I need to derive directly from ISourceBlock<T> rather than using the Encapsulate method?

Theodor Zoulias · Accepted Answer · 2022-12-22T11:24:40.123

You discovered a major disadvantage of the TPL Dataflow library: extending it is a pain. The interaction between the linked dataflow blocks is so convoluted that even Microsoft can't get it right, and there are still known bugs that are so hard to fix that have not been fixed, after being reported for years.

When you link the _source to the target in the SomeCustomBlock<T>.LinkTo method, the _source will start sending messages to the target directly, by invoking the target's ConsumeMessage method. Your SomeCustomBlock<T> class alone has no way of intercepting this communication between _source and target. What you can do is to write a ITargetBlock<T> implementation that will serve as a middle man between the _source and the target, whose activity you'll be able to intercept:

private class LinkedTargetProxy : ITargetBlock<T>
{
    private readonly ITargetBlock<T> _realTarget;
    private readonly SomeCustomBlock<T> _realSource;

    public LinkedTargetProxy(ITargetBlock<T> realTarget, SomeCustomBlock<T> realSource)
    {
        _realTarget = realTarget;
        _realSource = realSource;
    }
    //...
}

public IDisposable LinkTo(ITargetBlock<T> target, DataflowLinkOptions linkOptions)
{
    LinkedTargetProxy targetProxy = new(target, this);
    return _source.LinkTo(targetProxy, linkOptions);
}

You can find examples of using this technique in these answers:

Custom TPL Block: how to test Consume, Release, Reserve

1 Answers1