How to exclude all elements in Flux from another Flux

Question

I have two Fluxs one for successful elements another one holding the erroneous elements

Flux<String> success= Flux.just("Orange", "Apple", "Banana","Grape", "Strawberry");
Flux<String> erroneous = Flux.just("Banana","Grape",);

How can i filter the first Flux to execlude all the elements from the second one ?

Bidle

Answer 1

You may wish to consider collecting the Flux into a set, caching that set, and then using filterWhen as follows:

Mono<Set<String>> erroneousSet = erroneous.collect(Collectors.toSet()).cache();
Flux<String> filtered = success.filterWhen(v -> erroneousSet.map(s -> !s.contains(v)));

Gives:

Orange
Apple
Strawberry

This isn't the most concise solution (see below), but it enables the contents of erroneous to be cached. In this specific example that's a moot point, but if it's a real-world situation (not using Flux.just()) then erroneous could be recomputed on every subscription, and that could end up being incredibly (and unnecessarily) expensive in performance terms.

Alternatively, if the above really doesn't matter in your use case, filterWhen() and hasElement() can be used much more concisely as follows:

success.filterWhen(s -> erroneous.hasElement(s).map(x->!x))

Or with reactor-extra:

success.filterWhen(s -> BooleanUtils.not(erroneous.hasElement(s)))

Culhert

Just one word, Perfect – Bidle

Ideally there would be a hasNoElement() or similar to avoid the final map() call, but that's the best way around it for now :-) – Culhert

@MeladBasilius Just had a brainwave and updated with a slightly different alternative if you happen to have reactor-extra in place already. Not much point including it just for that though. – Culhert

note that using a Flux for that purpose can be a bit inadapted if you have the opportunity to first collect the elements to exclude in a set and use .filter(...) instead. especially if erroneous is recomputed everytime it is subscribed to, and the just is for demonstration purpose only – Murther

@SimonBaslé That's a good point. I guess if the Flux is the only available source (assuming it terminates), then a better option might be collecting as a set and caching, i.e. erroneousSet = erroneous.collect(Collectors.toSet()).cache() and success.filterWhen(v -> erroneousSet.map(s -> !s.contains(v))) - would that make sense or is there a more concise way I've missed? – Culhert

@SimonBaslé Thanks for the confirmation. I've rewritten the answer - I think it's all too easy to jump to the most concise solution otherwise without considering the performance impact (as I did when first writing that post!) – Culhert