Rxdart: Rework buffer and window

Created on 30 Mar 2018 · 2Comments · Source: ReactiveX/rxdart

Currently we have bufferCount/windowCount and bufferTime/windowTime, but we have no way to buffer on arbitrary intervals (for example, buffer on another Stream, or on a Future, ...)

Proposal:
We create a generic buffer/window operator which takes a StreamSampler implementation.
This sampler is a simple interface which handles how the buffer builds and when the buffer should be emitted.

We provide a set of samplers that are commonly used:

for count, [skip]: OnCount
for Duration: OnTime
for sampling on another Stream: OnStream
for a recurring Future: OnFuture
for a condition: OnTest

The user is free to create his/her own implementation OnX.

This will provide the following API:

    .buffer(onTime(const Duration(seconds: 1)))
   .listen(...);

observable
    .buffer(onFuture(() => window.animationFrame()))
   .listen(...);

// stream emits: 1, 5, 20, 80, 120, 150, 200
observable
    .buffer(onTest((int event) => event > 100))
   .listen(...); // prints [1, 5, 20, 80, 120], [150, 200]

Which is equivalent to the following shorthand:

    .bufferTime(const Duration(seconds: 1))
   .listen(...);

observable
    .bufferFuture(() => window.animationFrame())
   .listen(...);

// stream emits: 1, 5, 20, 80, 120, 150, 200
observable
    .bufferTest((int event) => event > 100)
   .listen(...); // prints [1, 5, 20, 80, 120], [150, 200]

in progress

Source