TPL Dataflow block to produce multiple outputs from a single input

Question

I have started looking into TPL Dataflow as a solution to a processing workflow.

The gist of the processing workflow is to read in input messages from multiple tables and create four reflecting objects from them and persist them to four other tables, so each input message should result in four new messages getting created.

I cannot identify one of the predefined blocks that can help with the creation of the four objects, at first TransformManyBlock seems to be what I am looking for but it returns multiple objects of the same type where I will have four types.

Example of the Problem

We have two tables containing Employee details from two legacy systems, their entities look like this

public partial class EmployeeTblA
{
    public int Id { get; set; }
    public int System { get; set; }
    public string Forename { get; set; }
    public string Surname { get; set; }
    public int Age { get; set; }
    public int Number { get; set; }
    public string Street { get; set; }
    public string PostCode { get; set; }

    public virtual EmployeeSystem SystemNavigation { get; set; }
}

public partial class EmployeeTblB
{
    public int Id { get; set; }
    public int System { get; set; }
    public string Name { get; set; }
    public int Age { get; set; }
    public string Address { get; set; }
    public string Postcode { get; set; }

    public virtual EmployeeSystem SystemNavigation { get; set; }
}

We want to take the data from the two systems and put the data into our shiny new system, to do this we need to convert the entities from the old system to the entities used in the new system. First we convert the entities from the old system to a base class which look like this

public class BaseEmployee
{
    public int Id { get; set; }
    public int System { get; set; }
    public string Name { get; set; }
    public int Age { get; set; }
    public string Address { get; set; }
    public string Postcode { get; set; }
}

We then want to create three new objects from the base class that represent the entities of the new system which looks like this

public partial class EmployeeName
{
    public int Id { get; set; }
    public int System { get; set; }
    public int LegacyId { get; set; }
    public string Name { get; set; }

    public virtual EmployeeSystem SystemNavigation { get; set; }
}

public partial class EmployeeAge
{
    public int Id { get; set; }
    public int System { get; set; }
    public int LegacyId { get; set; }
    public int Age { get; set; }

    public virtual EmployeeSystem SystemNavigation { get; set; }
}

public partial class EmployeeAddress
{
    public int Id { get; set; }
    public int System { get; set; }
    public int LegacyId { get; set; }
    public string Address { get; set; }
    public string Postcode { get; set; }

    public virtual EmployeeSystem SystemNavigation { get; set; }
}

The rough flow of my TPL for the above example

1. Read data from tables in DB into a TranformBlock<Employee,BaseEmployee> converting to a common object, this is down twice for each of the legacy systems

2. Each TranformBlock<Employee, BaseEmployee> is linked to a BatchBlock to group all the input flows.

3. BatchBlock is linked to a Block<BaseEmployee, ...> which will take the input and create two new objects from the input data, EmployeeName and EmployeeAge.

4. The Block<BaseEmployee, ...> will then be linked to Action block and Action which take save them to their respective table's in the DB

I know I can create a custom block but I cannot figure out how I could use it to provide the output to four separate linked ActionBlock using dataflow, can someone please point me in the right direction?

Answer 1

The Broadcast block was the component I ended up going with, I used it to broadcast the BaseEmployee object to other output streams splitting out the reflecting objects I needed to create.

Full pipeline same below

         _transEmployeeA = new TransformBlock<EmployeeTblA, BaseMsg>((input) =>
         {
            return new BaseMsg()
            {
                Id = input.Id,
                System = input.System,
                Name = string.Concat(input.Forename, " ", input.Surname),
                Age = input.Age,
                Address = string.Concat(input.Number, " ", input.Street),
                Postcode = input.PostCode
            };
        });

        _transEmployeeB = new TransformBlock<EmployeeTblB, BaseMsg>((input) =>
        {
            return new BaseMsg()
            {
                Id = input.Id,
                System = input.System,
                Name = input.Name,
                Age = input.Age,
                Address = input.Address,
                Postcode = input.Postcode
            };
        });

        _broadcastBaseMsg = new BroadcastBlock<BaseMsg>(null);

        _transEmployeeName = new TransformBlock<BaseMsg, EmployeeName>((baseMsg) =>
        {
            return new EmployeeName()
            {
                System = baseMsg.System,
                LegacyId = baseMsg.Id,
                Name = baseMsg.Name
            };
        });

        _transEmployeeAge = new TransformBlock<BaseMsg, EmployeeAge>((baseMsg) =>
        {
            return new EmployeeAge()
            {
                System = baseMsg.System,
                LegacyId = baseMsg.Id,
                Age = baseMsg.Age
            };
        });

        _transEmployeeAddress = new TransformBlock<BaseMsg, EmployeeAddress>((baseMsg) =>
        {
            return new EmployeeAddress()
            {
                System = baseMsg.System,
                LegacyId = baseMsg.Id,
                Address = baseMsg.Address,
                Postcode = baseMsg.Postcode
            };
        });


        _bufferEmployeeName = new BufferBlock<EmployeeName>();
        _bufferEmployeeAge = new BufferBlock<EmployeeAge>();
        _bufferEmployeeAddress = new BufferBlock<EmployeeAddress>();

        _actionEmployeeName = new ActionBlock<EmployeeName>((output) =>
        {
            using (var cxt = new SandboxContext())
            {
                cxt.EmployeeNames.Add(output);
                cxt.SaveChanges();
            }
        });

        _actionEmployeeAge = new ActionBlock<EmployeeAge>((output) =>
        {                
            using (var cxt = new SandboxContext())
            {
                cxt.EmployeeAges.Add(output);
                cxt.SaveChanges();
            }                
        });

        _actionEmployeeAddress = new ActionBlock<EmployeeAddress>((output) =>
        {                
            using (var cxt = new SandboxContext())
            {
                cxt.EmployeeAddresses.Add(output);
                cxt.SaveChanges();
            }               
        });

        var linkOpts = new DataflowLinkOptions()
        {
            PropagateCompletion = true
        };

        // Transform Employees and pass to Batch
        _transEmployeeA.LinkTo(_broadcastBaseMsg, linkOpts);
        _transEmployeeB.LinkTo(_broadcastBaseMsg, linkOpts);

        // Transform Broadcast to respective outputs
        _broadcastBaseMsg.LinkTo(_transEmployeeName, linkOpts);
        _broadcastBaseMsg.LinkTo(_transEmployeeAge, linkOpts);
        _broadcastBaseMsg.LinkTo(_transEmployeeAddress, linkOpts);

        // Add outputs to Buffer
        _transEmployeeName.LinkTo(_bufferEmployeeName, linkOpts);
        _transEmployeeAge.LinkTo(_bufferEmployeeAge, linkOpts);
        _transEmployeeAddress.LinkTo(_bufferEmployeeAddress, linkOpts);

        // Persist outputs to DB
        _bufferEmployeeName.LinkTo(_actionEmployeeName, linkOpts);
        _bufferEmployeeAge.LinkTo(_actionEmployeeAge, linkOpts);
        _bufferEmployeeAddress.LinkTo(_actionEmployeeAddress, linkOpts);

Additionally comments from @TheodorZoulias helped to simply my usage of TPL dataflow for this particular dataflow.