Rxjava - Group/Batch bursts of elements in an observable sequence

Question

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.

This sounds a little bit like the window() function. Have you checked if that meets your needs? — Austin Schaefer, Dec 12 '19 at 14:59
Isn't window exactly the same as buffer() but with observables as output instead? — Helv1e, Dec 12 '19 at 16:00

score 4 · Answer 1 · answered Dec 13 '19 at 18:03

window operator will act as buffer, you cannot use it directly.

The idea is to control the timer by the emissions of the first observable (which I call insertions). For this you have to include a third parameter to link the two observables (stopWatch Subject in the solution bellow).

    @Test
    public void stop_watch_observable() {

        Subject<Long> stopWatch = PublishSubject.create();

        Observable<Long> insertions = insertions();

        //share to use it as a timer (looking for the first emission)
        //and to recieve the items
        Observable<Long> shared = insertions.share();

        //for each emission of insertions we start a new timer
        //but only the first one is emitted
        //the others are stopped by the takeUntil(stopWatch)
        Observable<Long> window = shared
                .flatMap(e -> Observable.timer(3, TimeUnit.SECONDS).takeUntil(stopWatch));

        shared.buffer(window)
                //each time a window is generated we kill all the current timers
                .doOnNext(e -> stopWatch.onNext(0L))
                .blockingSubscribe(System.out::println);
    }

    // insertions generator which is comming randomly
    private Observable<Long> insertions() {
        AtomicLong al = new AtomicLong(0);
        return Observable.generate((Emitter<Long> emitter) -> {
            if (al.getAndIncrement() % 4 == 0) {
                Long timeToWait = Long.parseLong(RandomStringUtils.randomNumeric(1));
                System.out.println("sleeping for: " + timeToWait);
                sleep(timeToWait * 1000);
            } else {
                sleep(500);
            }
            emitter.onNext(al.get());
        }).subscribeOn(Schedulers.newThread());
    }

score 1 · Accepted Answer · answered Dec 17 '19 at 14:32

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

Rxjava - Group/Batch bursts of elements in an observable sequence

2 Answers2