using System;

using System.Reactive;

using System.Reactive.Concurrency;

using System.Reactive.Linq;

using System.Reactive.Subjects;

using System.Threading;

​

​

namespace ConsoleApplication6

{

    class Program

    {

        static void Main(string[] args)

        {

            DoStuff();

            Console.ReadKey();

        }

​

        static void DoStuff()

        {

            //TODO: generate Observers dynamically for each key 

            var aObserver = new Subject<Tuple<char, long>>();

            var bObserver = new Subject<Tuple<char, long>>();

​

            aObserver.ObserveOn(NewThreadScheduler.Default)

                    .Subscribe(i =>

                    {

                        Thread.Sleep(1000);

                        Console.WriteLine(i);

                    });

​

            bObserver.ObserveOn(NewThreadScheduler.Default)

                    .Subscribe(i =>

                    {

                        Thread.Sleep(1000);

                        Console.WriteLine(i);

                    });

​

            //Testdata

            var aSource = Observable.Interval(TimeSpan.FromMilliseconds(10))

                                    .Select(i => new Tuple<char,long>( 'a', i))

                                    .Take(10);

​

            var bSource = Observable.Interval(TimeSpan.FromMilliseconds(10))

                                    .Select(i => new Tuple<char, long>('b', i))

                                    .Take(10);

​

​

             aSource.Merge(bSource)//Merge test data as one Observable,

                  .GroupBy(i => i.Item1) // group by Key

                  .Subscribe(i => i.Subscribe(j =>

                  {

                      switch (i.Key) //There must be a better way to do this

                      {

                          case 'a':

                              aObserver.OnNext(j);

                              break;

                          case 'b':

                              bObserver.OnNext(j);

                              break;

                      }

                  }));

        }

    }

​

}

​