I have an observable sequence. When the first element is inserted, I would like to start a timer and batch subsequent inserted elements during the timespan of the timer. Then, the timer wouldn't start again until another element is inserted in the sequence.
--------|=====timespan====|---------------|=====timespan====|-------------->
1 2 3 4 5 6
would produce:
[1,2,3,4,5], [6]
I tried with Observable.buffer() and a timespan but from my experimentation, I can see that the timer is started as soon as we subscribe to the observable sequence and is restarted as soon as the previous timer is completed.
So having the same sequence as the previous example and using the buffer() with a timespan, I would have something like this:
|=====timespan====|=====timespan====|=====timespan====|=====timespan====|-->
1 2 3 4 5 6
which would produce this:
[1,2,3,4], [], [5,6], []
This is essentially the same question as 29858974, but for java instead.
So the problem is that since I don't want to delay my stream too much I want to have a timer that is quite short, and that timer would be quite intensive. I could simply filter empty lists, but I think that impacts the CPU too much.
The drawback of the first solution is that a timer is launched for each emission of insertions (it can be CPU intensive). Here another solution with only one timer started at time (I thinks it's more efficient this way:
@Test
public void stop_watch_observable() {
Observable<Long> insertions = insertions();
Observable<Long> shared = insertions.share();
AtomicBoolean timerOn = new AtomicBoolean(false);
Observable<Long> window = shared
.flatMap(e -> timerOn.get() ? Observable.empty() : Observable.timer(3, TimeUnit.SECONDS)
.doOnSubscribe(x -> timerOn.set(true))
);
shared.buffer(window)
//each time a window is generated we kill all the current timers
.doOnNext(e -> timerOn.set(false))
.blockingSubscribe(System.out::println);
}