c# - How can I alternately buffer and flow a live data stream in Rx

Question

I have two streams. One is a flow of data (could be any type), the other is a boolean stream acting as a gate. I need to combine these into a stream that has the following behaviour:

• When the gate is open (most recent value was true) then the data should flow straight through
• When the gate is closed (most recent value was false) then the data should be buffered to be released as individual elements when the gate is next open
• The solution should preserve all elements of the data and preserve order

I am not really sure how to put this together. The inputs I have been testing with are like this:

// 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);

Answer 1

I would do this as follows:

• Window the data stream using the gate stream as a closing selector
• We can use DistinctUntilChanged on the gate stream to ensure no repeated values
• We will also force the gate stream to start closed (false) - it won't affect the output and allows a neat trick
• Then use the overload of Select that gives each element an index number. With this we can tell if we need to buffer or just emit the window as is, because we know that even-numbered windows are for buffering (because we made sure the gate stream starts with false)
• We can use ToList() to buffer each even window until it closes - this is the actually equivalent of a Buffer() that waits until OnCompleted
• We use an identity SelectMany to flatten the buffered windows
• Finally we concatenate the windows in order to guarantee order is preserved

It looks like this:

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