How do I link multiple target blocks with a source block in TPL Dataflow?

Question

I expected the following to produce output from both publishers, but it only produces output from the first one:

var broadcastBlock = new BroadcastBlock<int>(null);
var transformBlock = new TransformBlock<int, int>(i => i*10);
var publish1 = new ActionBlock<int>(i => Console.WriteLine("Publisher 1:" + i));
var publish2 = new ActionBlock<int>(i => Console.WriteLine("Publisher 2:" + i));

broadcastBlock.LinkTo(transformBlock, new DataflowLinkOptions() { PropagateCompletion = true });
transformBlock.LinkTo(publish1, new DataflowLinkOptions() { PropagateCompletion = true });
transformBlock.LinkTo(publish2, new DataflowLinkOptions() { PropagateCompletion = true });

foreach (var i in Enumerable.Range(0, 5))
{
    broadcastBlock.Post(i);
}
broadcastBlock.Complete();
Task.WhenAll(publish1.Completion, publish2.Completion).Wait();

I'm obviously missing something fundamental here, any ideas?

Answer 1

You are linking 2 ActionBlocks to a single TransformBlock. You should be linking the 2 ActionBlocks to the BrodcastBlock and link the BroadcastBlock to the TransformBlock.

What you have:

BroadCast => Transfrom => ActionBlock
                       => ActionBlock

What you need:

Transfrom => BroadCast => ActionBlock
                       => ActionBlock

Answer 2

I can't comment above, but I thought my findings on BroadcastBlock would be useful to someone:

• As long as you are broadcasting to blocks that do NOT have a 'DataflowBlockOptions.BoundedCapacity' set, then ALL messages will be delivered.
• Make sure that your source block is the LAST one you link up. I discovered you can create a race condition if you link it first. The source and broadcasting blocks can process messages BEFORE the target blocks are linked. It would depend on the application, but in my trivial test, I would only see the last message in my target/action blocks.

Edit: noticing my second comment is not relevant to OP based on how they structured it, but something I was running into because my SourceBlock was being populated in a Task.