Handle stream data with Dataflow I have a websocket connection with 3-rd party API. The API returns a lot of data, in a peak hours it returns hundreds messages per second. I need to process the data for two purposes: save everything in a DB and send some data to RabbitMq.

The idea is the following:
I want to save data to DB when batch size is 1000 or by timeout which is equal to 3 seconds.    

I want to publish data to RabbitMQ by 1 second timeout. It's a kind of throttling, because there are a lot of data. Moreover, I need to select the last record for the specific ticker, I have done it in the ActionBlock, f.e: we have the following records in the batch:  


  {"Ticker": "MSFT", "DateTime": '2019-05-14T10:00:00:100'}  
  {"Ticker": "MSFT", "DateTime": '2019-05-14T10:00:00:150'} 
  {"Ticker": "AAPL", "DateTime": '2019-05-14T10:00:00:300'}


I need to publish the last for the specific ticker only, so after filtering there will be 2 records that I am going to publish:


  {"Ticker": "MSFT", "DateTime": '2019-05-14T10:00:00:150'} 
  {"Ticker": "AAPL", "DateTime": '2019-05-14T10:00:00:300'}


Full code:  

public class StreamMessagePipeline &lt; T &gt; where T: StreamingMessage {
    private readonly BatchBlock &lt; T &gt; _saveBatchBlock;
    private readonly BatchBlock &lt; T &gt; _publishBatchBlock;

    public StreamMessagePipeline() {
        _saveBatchBlock = new BatchBlock &lt; T &gt; (1000);
        _publishBatchBlock = new BatchBlock &lt; T &gt; (500);

        SetupSaveBatchPipeline();
        SetupPublishBatchPipeline();
    }

    private void SetupSaveBatchPipeline() {
        var saveBatchTimeOut = TimeSpan.FromSeconds(3);
        var saveBatchTimer = new Timer(saveBatchTimeOut.TotalMilliseconds);

        saveBatchTimer.Elapsed += (s, e) = &gt;_saveBatchBlock.TriggerBatch();

        var actionBlockSave = new ActionBlock &lt; IEnumerable &lt; T &gt;&gt; (x = &gt;{
            //Reset the timeout since we got a batch
            saveBatchTimer.Stop();
            saveBatchTimer.Start();

            Console.WriteLine($ "Save to DB : {x.Count()}");
        });

        _saveBatchBlock.LinkTo(actionBlockSave, new DataflowLinkOptions {
            PropagateCompletion = true
        });
    }

    private void SetupPublishBatchPipeline() {
        var publishBatchTimeOut = TimeSpan.FromSeconds(1);
        var publishBatchTimer = new Timer(publishBatchTimeOut.TotalMilliseconds);

        publishBatchTimer.Elapsed += (s, e) = &gt;_publishBatchBlock.TriggerBatch();

        var actionBlockPublic = new ActionBlock &lt; IEnumerable &lt; T &gt;&gt; (x = &gt;{
            var res = x.GroupBy(d =&gt; d.Ticker).Select(d = &gt;d.OrderByDescending(s =&gt;s.DateTime).FirstOrDefault()).ToList();

            Console.WriteLine($ "Publish data to somewhere : {res.Count()}");
            //Reset the timeout since we got a batch
            publishBatchTimer.Stop();
            publishBatchTimer.Start();

        });

        _publishBatchBlock.LinkTo(actionBlockPublic, new DataflowLinkOptions {
            PropagateCompletion = true
        });
    }

    public async Task Handle(T record) {
        await _saveBatchBlock.SendAsync(record);
        await _publishBatchBlock.SendAsync(record);
    }

}
General comments:  I would move the timeout and batch sizes into private fields.  I don't know what version of c# you are using but you could create a class or if you have the nice tuple support you can just create it like so

private readonly (TimeSpan timeout, int buffer) _saveBatchSettings = (TimeSpan.FromSeconds(3), 1000);
private readonly (TimeSpan timeout, int buffer) _publishBatchSettings = (TimeSpan.FromSeconds(1), 500);


It's nice to have these outside the main program because typically with batching you have to tweak the timeouts and size to find the sweet spot and this make it simpler to find what you need to change. 

Also you have no way to complete the source.  I would recommend making the class IDisposable and in the dispose mark the source as complete. 

TPL DataFlow comments:   Why not use a BroadcastBlock that is tied to the BatchBlocks then it's one send and not two.  

Move this

var res = x.GroupBy(d =&gt; d.Ticker).Select(d = &gt;d.OrderByDescending(s =&gt;s.DateTime).FirstOrDefault()).ToList();


Into a TransformBlock and make the ActionBlock just contain the code to publish the result.

For ActionBlock, this is just my preference, I make a method that the action block calls to instead of inline lambda it.  That code is typically the processing code and it's easier to maintain and read if it's in it's own method and outside the pipeline setup. 

ReactiveExtensions:  If you are willing to add ReactiveExtensions to your project you can mix and Rx and TPL DataFlow Blocks.  Rx has a built in method for your buffering called Buffer.  The DataFlow blocks has an AsObservable and AsObserver to switch to and from Rx and TPL. This would be an example for using Rx with TPL

public class StreamMessagePipeline&lt;T&gt; : IDisposable
    where T : StreamingMessage
{
    private BroadcastBlock&lt;T&gt; _source;

    private readonly (TimeSpan timeout, int buffer) _saveBatchSettings = (TimeSpan.FromSeconds(3), 1000);
    private readonly (TimeSpan timeout, int buffer) _publishBatchSettings = (TimeSpan.FromSeconds(1), 500);

    public StreamMessagePipeline()
    {
        _source = new BroadcastBlock&lt;T&gt;(x =&gt; x);
        SetupSaveBatchPipeline();
        SetupPublishBatchPipeline();
    }

    public async Task Handle(T record)
    {
        await _source.SendAsync(record);
    }

    private void SetupSaveBatchPipeline()
    {
        var actionBlockSave = new ActionBlock&lt;IList&lt;T&gt;&gt;(SaveBatch);

        // Instead of action block you could stay in Rx and just use Subscribe 
        _source.AsObservable()
            .Buffer(_saveBatchSettings.timeout, _saveBatchSettings.buffer)
            .Where(x =&gt; x.Count &gt; 0) // unlike TriggerBatch Buffer will send out an empty list
            .Subscribe(actionBlockSave.AsObserver());
    }

    private void SetupPublishBatchPipeline()
    {

        var transformBlock = new TransformBlock&lt;IList&lt;T&gt;, IList&lt;T&gt;&gt;(x =&gt;
        {
            return x.GroupBy(d =&gt; d.Ticker)
                .Select(d =&gt; d.OrderByDescending(s =&gt; s.DateTime).FirstOrDefault()).ToList();
        });

        var actionBlockPublish = new ActionBlock&lt;IList&lt;T&gt;&gt;(PublishBatch);

        transformBlock.LinkTo(actionBlockPublish, new DataflowLinkOptions()
        {
             PropagateCompletion = true,
        });

        _source.AsObservable()
            .Buffer(_publishBatchSettings.timeout, _publishBatchSettings.buffer)
            .Where(x =&gt; x.Count &gt; 0) // unlike TriggerBatch Buffer will send out an empty list
            .Subscribe(transformBlock.AsObserver());
    }

    private void SaveBatch(IList&lt;T&gt; messages)
    {
        // Save the batch
    }

    private void PublishBatch(IList&lt;T&gt; messages)
    {
        // publish the batch 
    }

    #region IDisposable Support
    private bool disposedValue = false; // To detect redundant calls

    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                _source.Complete();
            }
            disposedValue = true;
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }
    #endregion

}


Update If you don't want to use Rx you can still make a custom block to that has the same functionality.  

public static class TimerBatchBlock
{
    public static IPropagatorBlock&lt;T, T[]&gt; Create&lt;T&gt;(int batchSize, TimeSpan timeSpan, GroupingDataflowBlockOptions options = null)
    {
        var batchBlock = new BatchBlock&lt;T&gt;(batchSize, options ?? new GroupingDataflowBlockOptions());
        var broadCastBlock = new BroadcastBlock&lt;T[]&gt;(x =&gt; x);
        var bufferBlock = new BufferBlock&lt;T[]&gt;();

        // timer setup (System.Threading.Timer)
        var timer = new Timer(x =&gt; ((BatchBlock&lt;T&gt;)x).TriggerBatch(), batchBlock, timeSpan, timeSpan);
        var resetTimerBlock = new ActionBlock&lt;T[]&gt;(_ =&gt; timer.Change(timeSpan, timeSpan)); // reset timer each time buffer outputs
        resetTimerBlock.Completion.ContinueWith(_ =&gt; timer.Dispose());

        // link everything up
        var linkOptions = new DataflowLinkOptions()
        {
            PropagateCompletion = true,
        };
        broadCastBlock.LinkTo(resetTimerBlock, linkOptions);
        broadCastBlock.LinkTo(bufferBlock, linkOptions);
        batchBlock.LinkTo(broadCastBlock, linkOptions);

        return DataflowBlock.Encapsulate(batchBlock, bufferBlock);
    }
}


You would link BroadCastBlock -> TimerBatchBlock -> ActionBlock