Program.cs

using System;
using System.Reactive.Linq;
using System.Reactive.Subjects;

namespace RxWithhold
{
  class Program
  {
    [STAThread]
    static void Main(string[] args)
    {
      var data = Observable.Interval(TimeSpan.FromSeconds(1));
      var connection = new Subject<bool>();

      var connected = false;
      connection.Do(x => connected = x).Subscribe();

      data
        .Window(() => connection)
        .Subscribe(window =>
        {
          Console.Write("Starting new window: ");
          if (connected == false)
          {
            Console.WriteLine("Buffering until online");
            window
              .Buffer(() => connection.Where(x => x))
              .Subscribe(buffered => buffered
                                       .ToObservable()
                                       .Subscribe(ProcessNotification));
          }
          else
          {
            Console.WriteLine("Forwarding notifications");
            window.Subscribe(ProcessNotification);
          }
        });

      connection
        .Subscribe(x => Console.WriteLine("Connection changed: {0}", x));

      var enabled = true;
      while (true)
      {
        Console.ReadKey();

        connection.OnNext(enabled);
        enabled = !enabled;
      }
    }

    static void ProcessNotification(long notification)
    {
      Console.WriteLine(notification);
    }
  }
}