TPL DataFlow: creating a custom splitblock

Question

just need some help creating a custom splitblock with the help of the dataflow library which is part of the TPL in .Net.

All I want to achieve is a simple custom block that takes an input and splits this to multiple transformblocks. This is needed for filtering data, where I could log negativ entries and go on with the good one.

For my needs it should be enough to split the input to two different outputs. The class header should look something like this...

public abstract class SplitBlock<TInput, TOutputLeft, TOutputRight>

My problem is that I dont realy know how to go on. All I know is that I need two TransformBlocks:

var leftBlock  = new TransformBlock<TInput, TOutputLeft>(...)
var rightblock = new TransformBlock<TInput, TOutputRight>(...)

In all my tries I ended up having multiple ITargetBlocks for storing the input of the left and right block, but this cant be right, can it?

I appreciate every hint that you can give.

Answer 1

I would start with thinking about how the general interface of that class should look like. I think the simplest solution would be something like:

public class SplitBlock<TInput, TOutputLeft, TOutputRight>
{
    public ITargetBlock<TInput> Input { get; }
    public ISourceBlock<TOutputLeft> LeftOutput { get; }
    public ISourceBlock<TOutputRight> RightOutput { get; }
}

With that, the implementation follows naturally: one input block connected to two output blocks. The only question is whether the actual processing should be done in the output blocks (like you suggested with your two TransformBlocks) or in the input block.

If you want to have processing in the output blocks, the input block can be an ActionBlock that sends the input to both outputs and the outputs will be TransformBlocks, as you suggested.

public class SplitBlock<TInput, TOutputLeft, TOutputRight>
{
    private ActionBlock<TInput> input;
    private TransformBlock<TInput, TOutputLeft> leftOutput;
    private TransformBlock<TInput, TOutputRight> rightOutput;

    public ITargetBlock<TInput> Input { get { return input; } }
    public ISourceBlock<TOutputLeft> LeftOutput { get { return leftOutput; } }
    public ISourceBlock<TOutputRight> RightOutput { get { return rightOutput; } }

    public SplitBlock(
        Func<TInput, TOutputLeft> leftTransform,
        Func<TInput, TOutputRight> rightTransform)
    {
        input = new ActionBlock<TInput>(
            x =>
            {
                leftOutput.Post(x);
                rightOutput.Post(x);
            });
        leftOutput = new TransformBlock<TInput, TOutputLeft>(leftTransform);
        rightOutput = new TransformBlock<TInput, TOutputRight>(rightTransform);

        // TODO handle fault in input correctly
        input.Completion.ContinueWith(
            _ =>
            {
                leftOutput.Complete();
                rightOutput.Complete();
            });
    }
}

(This assumes that left and right transforms can process the same input at the same time.)

On the other hand, if you wanted to perform the processing in the input block (which makes more sense to me), then you could have ActionBlock as input and BufferBlocks as outputs, with the input block processing the input and then sending the results to the output blocks:

public class SplitBlock<TInput, TOutputLeft, TOutputRight>
{
    private ActionBlock<TInput> input;
    private BufferBlock<TOutputLeft> leftOutput;
    private BufferBlock<TOutputRight> rightOutput;

    public ITargetBlock<TInput> Input { get { return input; } }
    public ISourceBlock<TOutputLeft> LeftOutput { get { return leftOutput; } }
    public ISourceBlock<TOutputRight> RightOutput { get { return rightOutput; } }

    public SplitBlock(
        Func<TInput, Tuple<TOutputLeft, TOutputRight>> combinedTransform)
    {
        input = new ActionBlock<TInput>(
            value =>
            {
                var result = combinedTransform(value);
                leftOutput.Post(result.Item1);
                rightOutput.Post(result.Item2);
            });
        leftOutput = new BufferBlock<TOutputLeft>();
        rightOutput = new BufferBlock<TOutputRight>();

        // TODO handle fault in input correctly
        input.Completion.ContinueWith(
            _ =>
            {
                leftOutput.Complete();
                rightOutput.Complete();
            });
    }

    public SplitBlock(
        Func<TInput, TOutputLeft> leftTransform,
        Func<TInput, TOutputRight> rightTransform)
        : this(x => Tuple.Create(leftTransform(x), rightTransform(x)))
    {}
}