Buffers values from one observable, by another 'gate' observable. Allows flows to be turned on an off without loosing intermediate values.

BufferByGate.cs

//From James World http://stackoverflow.com/questions/23431018/how-can-i-alternately-buffer-and-flow-a-live-data-stream-in-rx

// a demo data stream that emits every second
var dataStream = Observable.Interval(TimeSpan.FromSeconds(1));

// a demo flag stream that toggles every 5 seconds
var toggle = false;
var gateStream = Observable.Interval(TimeSpan.FromSeconds(5))
    .Select(_ => toggle = !toggle);

dataStream.Window(gateStream.StartWith(false).DistinctUntilChanged())
          .Select((w, i) =>  i % 2 == 0 ? w.ToList().SelectMany(x => x) : w)
          .Concat()
          .Subscribe(Console.WriteLine);