Reactor-core: Add a .filter(Function<T, Mono<Bool>>) operator

[nebhale]

This was brought up in https://github.com/cloudfoundry/cf-java-client/issues/715 a couple of days ago and I think warrants thought. Currently, there is no clean way to take a value that you want to filter() on, do some asynchronous work on it, and return the result of thefilter(). A work around is described in the Cloud Foundry Java Client issue, but looks something like:

    .just("a", "b", "c")
    .then(value -> asyncOperation(value)
        .and(Mono.just(value))    
    .filter(function((asyncValue, value) -> asyncValue == "foo"))
    .map(function((asyncValue, value) -> value))
    .subscribe(System.out::println)

Effectively, you have to attach the original value as "context" (via a Tuple) to the value you _actually_ want to filter() on, perform the filtering, and then extract that original value from the "context". I'd instead prefer to see:

    .just("a", "b", "c")
    .filter(value -> asyncOperation(value)
        .map(asyncValue -> asyncValue == "foo"))
    .subscribe(System.out::println)

I believe that it both reads better and communicates the intent of the flow better.

[akarnokd]

FYI: https://github.com/akarnokd/RxJava2Extensions#flowabletransformersfilterasync

https://github.com/akarnokd/RxJava2Extensions/blob/master/src/main/java/hu/akarnokd/rxjava2/operators/FlowableFilterAsync.java

[simonbasle]

we shall name it... filterWhen. Similar to the rx-extras operator linked by David, an empty Mono will be considered "false" for the filter purpose (rejecting the value), and an error will be propagated.

[nebhale]

An empty Mono will also be considered false or is it filterWhen(Function<T, Mono<T>>)?

[simonbasle]

@nebhale An empty Mono<Boolean> will also be considered false for the purpose of the filter.

[smaldini]

Not sure if we should enforce Mono, When like Other suffix tends to pick up first signal and cancel (which is annoying because of #442).

[simonbasle]

@smaldini sure, we can accept Publisher<Boolean> and only consider the first emitted boolean for the filter, then cancel if not a Mono.

[nebhale]

Sounds good. Leaves a bit of flexibility for you to use .map() or .filter() as part of the predicate flow.