using System;

using System.Collections.Generic;

using System.Collections.Specialized;

using System.Diagnostics;

using System.IO;

using System.Linq;

using System.Net;

using System.Reactive;

using System.Reactive.Concurrency;

using System.Reactive.Linq;

using System.Reactive.Subjects;

using System.Text;

using System.Threading;

​

public class Program

{

    public static void Main()

    {

        var obs = new Subject<int>();

        var publications = 

            obs

            .Do(x => Console.WriteLine("to buffer {0}", x), () => Console.WriteLine("to buffer complete"))

            .Buffer(2)

            .Do(x => Console.WriteLine("from buffer {0}", ShowContent(x)), () => Console.WriteLine("from buffer complete"))

            .ObserveOn(ThreadPoolScheduler.Instance)

            .Do(x => Console.WriteLine("to selectmany {0}", ShowContent(x)), () => Console.WriteLine("to selectmany complete"))

            .SelectMany(x => Test(x).ToEnumerable())

            .Do(x => Console.WriteLine("notify {0}", x), () => Console.WriteLine("complete"));

        publications

            .Do(x => Console.WriteLine("publications notify {0}", x), () => Console.WriteLine("publications complete"))

            .Subscribe()

            ;

        obs.OnNext(1);

        obs.OnNext(2);

        obs.OnNext(3);

        var nextpub = publications.FirstAsync();

        obs.OnCompleted();

        nextpub.Wait();

    }

    static string ShowContent<T>(IEnumerable<T> e) {

        return "[" + string.Join(", ", e.Select(x => x.ToString())) + "]";

    }

    static IObservable<Unit> Test(IList<int> _) {

        Console.WriteLine("> thread: {0}", Thread.CurrentThread.ManagedThreadId);

        Thread.Sleep(50);

        return Observable.Return(Unit.Default);

    }

}