using System;

using System.Collections.Generic;

using System.Linq;

using System.Reactive.Linq;

using System.Threading.Tasks;

​

public static class Program

{

    public static async Task Main()

    {   

        var syncedProducers =

            // ConnectedProducersEvent ticks an array of connected producers, each time a producer connects or disconnects

            ConnectedProducersEvent

                .SelectMany(producers =>

                    Observable

                        .Merge(producers) // Put all objects, from all producers into the same observable

                        .GroupBy(@object => @object.SSId) // Group objects by matching SSId

​

                    .SelectMany(group => group.Buffer(producers.Length))); // Syncing: Emit the SSId group, when the group count matches the count of connected producers

​

        // Now you can wire syncedProducers to consumers

        var consumer1 =

            syncedProducers

                .Select(x => x.Where(y => y.ProducerId == 1));

​

        await syncedProducers

            .Do(WriteToConsole);

    }

​

    private static void WriteToConsole(IList<Product> syncedProducts)

    {

        Console.WriteLine($"Synced {syncedProducts.Count} producers on SSID {syncedProducts.First().SSId}");

        Console.WriteLine();

    }

​

    private static Random _rnd = new Random();

​

    public class Product

    {

        public int ProducerId { get; set; }

        public long SSId { get; set; }

    }

​

    private static IObservable<IObservable<Product>[]> _connectedProducersEvent;

​

    // Create three events, with 2, then 3 and finally 4 producers

    // Each producer create 3 'products' at random times

    // Each producer writes to console

    private static IObservable<IObservable<Product>[]> ConnectedProducersEvent

    {

        get

        {

            if (_connectedProducersEvent == null)

            {

                _connectedProducersEvent =

                    Observable

                        .Interval(TimeSpan.FromMilliseconds(300))

                        .Take(3)// Generate three events

                        .Select(x => (int)x + 2) // Generate 2, then 3 and finally 4 producers

                        .Do(x => Console.WriteLine($"***** ConnectedProducersEvent! Count: {x} *****"))

                        .Select(producerCount =>

                            Enumerable

                                .Range(0, producerCount)

                                .Select(producerId =>

                                {

                                    return Observable

                                        .Interval(TimeSpan.FromMilliseconds(_rnd.Next(1, 100)))

                                        .Select(ssid => new Product { ProducerId = producerId, SSId = ssid })

                                        .Do(x => Console.WriteLine($"P{x.ProducerId}: SSId:{x.SSId}"))

                                        .Take(3);

                                })

                                .ToArray());

            }

            return _connectedProducersEvent;

        }

    }

}