CODEWITHSUNDEEP
c#system.reactive
Lee F
Lee F

Reputation: 569

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

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:

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

Upvotes: 10

Views: 1201

Answers (1)

James World
James World

Reputation: 29776

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

Upvotes: 17

Related Questions