How to retain elements of Flux that filtered out

user_12225145 0
Viewed 387

I'm using reactor-core:3.3.0.RELEASE and want to retain some elements of Flux that filtered out by a Predicate.

I just came up with the following code, however, I know it's not efficient and looks awkward because it applies Predicate twice.

Queue<Integer> queue = Queues.<Integer> small().get();
Predicate<Integer> evenNumberPredicate = i -> i % 2 == 0;

Flux.range(1, 50)
    .map(i -> {
        if (!evenNumberPredicate.test(i)) {
            queue.offer(i);  // odd numbers are queued: want to use them later
        }
        return i;
    })
    .filter(evenNumberPredicate)
    .subscribe(System.out::println); // even numbers are printed

I'd like to code something like this;

Flux.range(1, 50)
    .filterOr(evenNumberPredicate, i -> queue.offer(i))
    .subscribe(System.out::println);

Any ideas? thank you.

project-reactor java
2 Answers

You could do it with groupBy operator.

Flux.range(1, 50)
        .groupBy(i -> i % 2)
        .flatMap(groupedFlux -> groupedFlux.key() == 0 ?
                groupedFlux :
                groupedFlux.doOnNext(queue::offer).ignoreElements())
        .subscribe(System.out::println);
user_102644300

I think in this case I'd just use a single ternary flatMap() call:

Flux.range(1, 50)
        .flatMap(i -> i%2==0 ?
                Mono.just(i) :
                Mono.just(i).doOnNext(queue::offer).ignoreElement())
        .subscribe(System.out::println);

Either way, ignoreElement() inside a flatMap() call is your friend here.

user_5514060
Related