2.x : window(timespan, unit, count) always emit empty observable if not reach max count

[xiemeilong]

This code will print nothing all the time.

PublishSubject<String> vehicleToFetch = PublishSubject.create();
        vehicleToFetch
                .delay(2,TimeUnit.SECONDS)
                .window(10, TimeUnit.SECONDS, 5)
                .observeOn(Schedulers.io())
                .subscribe(w-> w.toList().subscribe(ws-> {
                    ws.forEach(v -> {
                        System.out.println(String.format("%s %d", v, Thread.currentThread().getId()));
                        vehicleToFetch.onNext(v);
                    });
                }));


        Observable.just("v1","v2","v3","v4")
                .subscribe(v->{
                    vehicleToFetch.onNext(v);
                });

I am using rxjava:2.0.6.

[akarnokd]

There are two problems with your code:

• You are using toList which requires a finite stream. Since you don't call onComplete on vehicleToFetch the sequence above won't print anything.
• You are probably not waiting long enough to see the results. The default RxJava schedulers are daemon threads and when the "main" thread ends, the whole application stops.

PublishSubject<String> vehicleToFetch = PublishSubject.create();
vehicleToFetch
        .delay(2,TimeUnit.SECONDS)
        .window(10, TimeUnit.SECONDS, 5)
        .observeOn(Schedulers.io())
        .subscribe(w-> w.toList().subscribe(ws-> {
            ws.forEach(v -> {
                System.out.println(String.format("%s %d", v, Thread.currentThread().getId()));
                vehicleToFetch.onNext(v);
            });
        }));


Observable.just("v1","v2","v3","v4").subscribe(vehicleToFetch);

Thread.sleep(3000);

[xiemeilong]

@akarnokd Thanks. I am calling toList() on a window observable ,not on vehicleToFetch, There is print if I change .window(10, TimeUnit.SECONDS, 5) to .window(10, TimeUnit.SECONDS, 4).

I am using a CountDownLatch(not show in code) to wait a long time.

[akarnokd]

Yes, because the window with limit 4 will call onComplete for you on the window that toList() consumes. With 5 there is noone to call onComplete. If you had a "v5" in that case, you'd see the printout again.

[xiemeilong]

Why did it not call onComplete when the timeout is reached? Does not this method mean either timeout or max count is reach will emit a window?

[akarnokd]

Oh, I see it now. There is a bug in the operator that doesn't complete the old window if a new one is due to the timeout. I'll post a fix for it.

[akarnokd]

Closing via #5106.

[xiemeilong]

It worked after upgrade to 2.0.7, but after three times timeout, there is no timeout anymore.

[akarnokd]

Do you have a new unit test for it?

[xiemeilong]

You can use the same code , it will block after three new window emitted.

[akarnokd]

Odd. I'll look into it again.

[akarnokd]

Found a couple of remaining issues. See PR #5213; the example works with it properly for me.