using System;

using System.Reactive.Linq;

using System.Reactive.Subjects;

​

namespace ConsoleApplication12

{

    public class Program

    {

        public static void Main(string[] args)

        {

            var values = new Subject<int>();

            var pauser = new Subject<bool>();

​

            var subscription = values.Pausable(pauser).Subscribe(v => Console.WriteLine(v));

​

            values.OnNext(0);

            pauser.OnNext(false);

            values.OnNext(1);

            values.OnNext(2);

            pauser.OnNext(true);

            values.OnNext(3);

            values.OnNext(4);

            values.OnNext(5);

            pauser.OnNext(false);

            values.OnNext(6);

            pauser.OnNext(true);

            values.OnNext(7);

​

            subscription.Dispose();

​

            pauser.OnNext(false);

            values.OnNext(8);

            pauser.OnNext(true);

            values.OnNext(9);

        }

    }

​

    static class ObservableExtensions

    {

        public static IObservable<T> Pausable<T>(this IObservable<T> source, IObservable<bool> pauser)

        {

            var queue = source.Buffer(pauser.Where(toPause => !toPause),

                                      _ => pauser.Where(toPause => toPause))

                              .SelectMany(l => l.ToObservable().Do(_ => Console.Write("from queue: ")));

            return source.Window(pauser.Where(toPause => toPause).StartWith(true), 

                                 _ => pauser.Where(toPause => !toPause))

                         .Switch()

                         .Merge(queue);

        }

    }

}

​