using System;

using System.Collections.Generic;

using System.Linq;

using System.Reactive;

using System.Reactive.Linq;

using System.Reactive.Subjects;

using System.Threading.Tasks;

​

public class Program

{

    private static Subject<string> _queue = new Subject<string>();

    public static int Main()

    {       

        // Shared method for each queue to handle the data row and bulk insert

        var insertObserver = new AnonymousObserver<IList<string>>(x => {

            Console.WriteLine($"Generic Bulk insert: {(string.Join(",", x))}");

        });

        var numberQueue = new Subject<string>();

        var addressQueue = new Subject<string>();

        var entityQueue = _queue

            .Where(x => x[0] == 'a')

            // Pretend doing some work on entity

            .Do(x => {

                numberQueue.OnNext($"{x}-Entity-Number-1");

                numberQueue.OnNext($"{x}-Entity-Number-2");

                addressQueue.OnNext($"{x}-Entity-Address-1");

                addressQueue.OnNext($"{x}-Entity-Address-2");

                addressQueue.OnNext($"{x}-Entity-Address-3");

            })

            .Select(x => $"{x}-Entity")

            //.Do(x => Console.WriteLine(x))

            .Buffer(5);

        // Pretend this is the queue for maters

        var matterQueue = _queue.Where(x => x[0] == 'b').Select(x => $"{x}-Matter").Buffer(5);

        var workItemQueue = _queue.Where(x => x[0] == 'c').Select(x => $"{x}-WorkItem").Buffer(5);

        var otherQueue = _queue.Where(x => x[0] != 'a' && x[0] != 'b' && x[0] != 'c')

            .Select(x => $"{x}-Other")

            .Buffer(5);

        entityQueue.Subscribe(insertObserver);

        matterQueue.Subscribe(insertObserver);

        workItemQueue.Subscribe(insertObserver);

        otherQueue.Subscribe(insertObserver);

        numberQueue.Buffer(5).Subscribe(insertObserver);

        addressQueue.Buffer(5).Subscribe(insertObserver);

        foreach (var data in GetData().Take(500).ToList()) {

            // Queue up the data for processing 

            _queue.OnNext(data);

        }

        _queue.OnCompleted();

        return 0;       

    }

    /// Pretend this is the JSON data

    public static IEnumerable<string> GetData() {       

        while(true)

        { 

            yield return Guid.NewGuid().ToString("n").Substring(0, 6);

        }

    }

}

​

​