Rewriting to single pipeline by using data flow blocks I have a number of messages with a payload as a list of items. My goal is to transform the message payload without losing the relation between the message and payloads.

Isn't it possible to rewrite the code to a single pipeline without creating an instance of a nested pipeline for each message?

using System.Threading.Tasks.Dataflow;

public class Message&lt;T&gt;
{
    public int Id { get; set; }

    public IList&lt;T&gt; Items { get; set; }
}

class Program
{
    static void Main(string[] args)
    {
        var rnd = new Random();

        var input = new TransformBlock&lt;Message&lt;int&gt;, Message&lt;string&gt;&gt;(x =&gt; Transform(x));

        var action = new ActionBlock&lt;Message&lt;string&gt;&gt;(
            x =&gt;
            {
                Console.WriteLine("MSG #{0}", x.Id);

                foreach (var item in x.Items)
                {
                    Console.WriteLine("\t {0}", item);
                }                    
            });


        input.LinkTo(action, new DataflowLinkOptions { PropagateCompletion = true });

        for (var idx = 1; idx &lt;= 3; idx++)
        {
            var msg = new Message&lt;int&gt; { Id = idx, Items = Enumerable.Range(0, rnd.Next(0, 10)).ToList() };
            input.Post(msg);
        }

        input.Complete();
        action.Completion.Wait();

        Console.WriteLine("Done");
        Console.ReadKey();
    }

    private static async Task&lt;Message&lt;string&gt;&gt; Transform(Message&lt;int&gt; input)
    {
        var linkOpts = new DataflowLinkOptions { PropagateCompletion = true };

        var itr = new TransformManyBlock&lt;IList&lt;int&gt;, int&gt;(x =&gt; x);
        var multiply = new TransformBlock&lt;int, int&gt;(x =&gt; x * 10);
        var convert = new TransformBlock&lt;int, string&gt;(x =&gt; "STR-" + x);

        itr.LinkTo(multiply, linkOpts);
        multiply.LinkTo(convert, linkOpts);

        var pipeline = DataflowBlock.Encapsulate(itr, convert);

        pipeline.Post(input.Items);
        pipeline.Complete();

        var buffer = new BufferBlock&lt;string&gt;();
        pipeline.LinkTo(buffer);

        await pipeline.Completion;

        IList&lt;string&gt; items;
        if (!buffer.TryReceiveAll(out items))
        {
            items = new List&lt;string&gt;();
        }

        return new Message&lt;string&gt; { Id = input.Id, Items = items };
    }
}
Isn't it possible to rewrite the code to a single pipeline without
  creating an instance of a nested pipeline for each message?


It really seems a hard task because you are putting the pipeline building logic into the transform Method. Create a method to build your pipeline, it is really asking for it!

private static PipelineType _pipeline;
private static PipelineType BuildPipeline(Message&lt;int&gt; input){
    if(_pipeline != null){
        _pipeline.Post(input.Items);
        _pipeline.Complete();
        return _pipeline;
    }
    var linkOpts = new DataflowLinkOptions { PropagateCompletion = true };

    var itr = new TransformManyBlock&lt;IList&lt;int&gt;, int&gt;(x =&gt; x);
    var multiply = new TransformBlock&lt;int, int&gt;(x =&gt; x * 10);
    var convert = new TransformBlock&lt;int, string&gt;(x =&gt; "STR-" + x);

    itr.LinkTo(multiply, linkOpts);
    multiply.LinkTo(convert, linkOpts);

    _pipeline = DataflowBlock.Encapsulate(itr, convert);

    _pipeline.Post(input.Items);
    _pipeline.Complete();

    var buffer = new BufferBlock&lt;string&gt;();
    _pipeline.LinkTo(buffer);

    return _pipeline;
}


To be able to acess the buffer in your transform method you should set it public. But actually the best option would be that await pipeline.Completion would return the result you need.