Reactor-core: filterWhen + flatMap makes the flux stop to emit

[rotilho]

I have the following flux:

@StreamListener(Sink.INPUT)
    public void listenEvent(Flux<Message<Event>> flux) {
        flux
                .doOnNext(System.out::println)
                .filterWhen(o1 -> Mono.just(false)) //REMOVE
                .flatMap(s -> Flux.just(new Object())) //REMOVE
                .subscribe();
    }

After process a bunch of message the flux just stop to emit new messages. I debugged Spring Cloud Streams and everything is fine from it perspective, the application continue to push new messages using Flux.create(emitter ...).

Today I realized that if I remove .filterWhen(o1 -> Mono.just(false)) and keep the .flatMap(s -> Flux.just(new Object())) it will work fine (the opposite is also true).

Am I doing something wrong or is it a bug?

_(There are a also a case connected with publish(flux) as well but I couldn't reproduce it yet in a short version)_

_Reactor Core 3.0.7.RELEASE and 3.1.0.M2_

[rotilho]

Without Spring Cloud Streams:

   @Test
    public void test() throws Exception {
        Flux<Object> flux = Flux.create(emitter -> {
            while (true) {
                emitter.next(new Object());
            }
        }).publish().autoConnect();

        flux
                .doOnNext(System.out::println)
                .filterWhen(o1 -> Mono.just(false)) //REMOVE
                .flatMap(m -> Flux.just(new Object())) //REMOVE
                .subscribe();
    }

[akarnokd]

.filterWhen(o1 -> Mono.just(false)) is equal to filter(o1 -> false) thus it filters out all elements.

[rotilho]

It's just an example to reproduce the problem. The issue here is that the flux stop to emit new objects if I use filterWhen + flatMap.

If you run the code you are going to see that doOnNext will stop to print objects after a while however if you comment out filterWhen or flatMap it will work fine.

[akarnokd]

Standalone unit test. It first requests 256 and then 192 => count == 448.

@Test
public void filterAllOut() {
    int[] count = { 0 };
    Flux.range(1, 1000)
    .doOnNext(v -> count[0]++)
    .doOnRequest(System.out::println)
    .filterWhen(o1 -> Mono.just(false)) //REMOVE
    .flatMap(s -> Flux.just(new Object())) //REMOVE
    .subscribe();

    Assert.assertEquals(1000, count[0]);
}

[rotilho]

Is it a issue or am I missing something? I didn't catch the problem from the implementation.

Btw, if I use filter(o1 -> false) instead of filterWhen(o1 -> Mono.just(false)) it works.

[akarnokd]

It's a bug in the operator: https://github.com/akarnokd/RxJava2Extensions/commit/b5ed392bef4f96863cb31e998c9d1467acf86bf1

Using the consumerIndex as the indicator for the emission was wrong, they have to be accounted separately.

[simonbasle]

cheers @akarnokd, I'll port your fix to reactor ;)

[simonbasle]

cc @osi, who was thinking about porting mapAsync, also impacted by this bug on RxJava2Extensions side

[smaldini]

@simonbasle I'm not sure mapAsync is required yet as its easy, although with overhead, to do flatMap( xxx.next()) to achieve the same if i get the intent of mapAsync correctly.

[JakubParzonka]

Hello, is it fixed? I just want to use same construction.

[simonbasle]

@JakubParzonka yes it is fixed as indicated by the referenced commit above, from almost 3 years ago

[billydh]

is there a simple answer to what the difference is between filterWhen vs filter and when one should use which?

[simonbasle]

yes: filter is for synchronous Predicates that run instantly, while filterWhen is for asynchronous predicates represented as Mono<Boolean> (or less typically Publisher<Boolean>, in which case only the first boolean is considered)