DataflowPipelineBuilder

Build TPL Dataflow pipelines using a fluent API and helpful extensions

Rationale

TPL Dataflow is a powerful library, but the code is far less readable without the help of a diagram, this library should help visualize the pipeline in code, too. See the last example (Simple extensions) for the pipeline in code that reseables a diagram.

NuGet

https://www.nuget.org/packages/DataflowPipelineBuilder

Using LINQ-like extension syntax

varbuilder=newBuilder();varpipelineBlock=builder.FromEnumerable(Enumerable.Repeat("{ }",10)).Select(JObject.Parse).Batch(3).SelectMany().Select(j =>j).Batch(3).End();varactual=awaitpipelineBlock.ReceiveAllAsync();Assert.That(actual.Count,Is.EqualTo(4));Assert.That(actual.Count(b =>b.Count==3),Is.EqualTo(3));Assert.That(actual.Count(b =>b.Count==1),Is.EqualTo(1));

Using blocks

varbuilder=newBuilder();varpipelineBlock=builder.Create(newBufferBlock<string>()).Then(newTransformBlock<string,JObject>(json =>JObject.Parse(json))).Then(newBatchBlock<JObject>(3)).End();awaitTask.WhenAll(Enumerable.Repeat("{ }",10).Select(pipelineBlock.SendAsync));pipelineBlock.Complete();varactual=awaitpipelineBlock.ReceiveAllAsync();Assert.That(actual.Count,Is.EqualTo(4));Assert.That(actual.Count(b =>b.Length==3),Is.EqualTo(3));Assert.That(actual.Count(b =>b.Length==1),Is.EqualTo(1));

Fork and join

varbuilder=newBuilder();varpipelineBlock=builder.Create<string>().Select(JObject.Parse).Fork(source =>source.Select(s =>s.Properties().First().Name),
source =>source.Select(s =>s.Properties().First().Value.Value<int>())).End();awaitEnumerable.Range(1,3).Select(index =>$@"{{ ""value{index}"": {index} }}").Select(pipelineBlock.SendAsync).WhenAll().Then(()=>pipelineBlock.Complete());varactual=awaitpipelineBlock.ReceiveAllAsync();Assert.That(actual.Count,Is.EqualTo(3));Assert.That(actual.Select(t =>t.Item1),Is.EquivalentTo(new[]{"value1","value2","value3"}));Assert.That(actual.Select(t =>t.Item2),Is.EquivalentTo(new[]{1,2,3}));

Simple extensions

Alternatively, the project SimpleDataflowPipelineBuilder contains extension methods applying to the dataflow blocks directly without the help of a builder (this doesn't support added logging and custom options applied to all links, but it's simple and works well in most cases).

varinput=BufferBlock(token);varoutput=input.Then(ParseDataBlock()).ForkJoinThen(source =>source.Then(BatchBlock()).Then(ApiCallToExternalResourceBlock()).Then(SelectManyBlock()),
source =>source.Then(ApiCallToExternalResourceThatDoesNotAcceptBatches())).Then(JoinResultsBlock()).Then(WriteToStorageBlock());