Reactor's Flux Fan out and Zip it?

user_656592 0
Viewed 234

I am trying to do following operations on Flux/Publisher which can only be read once ( think database results which can be read once). But, this question is generic enough that it can be answered in functional programming context without reactor knowledge.

  1. count unique item
  2. check if an element exists
  3. Don't call the publisher/flux generator multiple times.
distinctAndHasElement(4, Flux.just(1,2,3,3,4,4,5));


Mono<Pair<Long, Boolean>> distinctAndHasElement(int toCheck, Flux<Integer> intsFlux) {
   // Code that doesn't work, Due to use of non final local variable
   boolean found = false;
   return intsFlux.map(x -> {
            if (toCheck == x) {
                found = true;
            }
            return x;
       })
       .distinct()
       .count()
       .map(x -> Pair.of(x, found));
}

We just need ability to fan out into 2 functions that operate on the same type/domain, and zip the final result.

Following doesn't work due to constrain#3

Flux<Integer> distinct = intsFlux.distinct();
Mono<Boolean> found = distinct.hasElement(toCheck);
Mono<Long> count = distinct.count();
return Mono.zip(count, found);
project-reactor functional-programming java
2 Answers

What you're attempting to do is a reduction of your dataset. It means that you attempt to create a single result by merging your initial elements.

Note that count can be considered as a kind of reduction, and in your case, you want an advanced kind of count operation, that also check if at least one of the input elements is equal to a given value.

With reactor (and many other stream framework), you can use the reduce operator.

Let's try your first example with it :

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

public class CountAndCheck {
    static Mono<Tuple2<Long, Boolean>> distinctAndHasElement(int toCheck, Flux<Integer> intsFlux) {
        return intsFlux
                .distinct()
                .reduce(Tuples.of(0L, false), (intermediateResult, nextElement) -> {
                    return Tuples.of(intermediateResult.getT1() + 1L, intermediateResult.getT2() ||  toCheck == nextElement);
                });
    }

    public static void main(String[] args) {
        System.out.println(distinctAndHasElement(2, Flux.just(1, 2, 2, 3, 4, 4)).block());
    }
}

The above program prints: [4,true]

Note: You can use the scan operator instead of reduction, to get a flux of every intermediate step in the reduction operation. It can be useful to understand how reduction is performed.

user_26780970

You can broadcast your Flux as described in the documentation.

Flux<Integer> distinct = intsFlux.distinct().publish().autoConnect(2);
Mono<Boolean> found = distinct.hasElement(toCheck);
Mono<Long> count = distinct.count();
return Mono.zip(count, found);
user_60511760
Related