Какой правильный способ архивации Flux с Mono?

Question

Существует Mono<A> и Flux<B>, и нам нужно создать поток кортежей, подобных этому:

Mono<A> monoA = createMono(); // {a}
Flux<B> fluxB = createFlux(); // {b1, b2, ... b100, ...}

Flux<Tuple<A,B>> zippedTuples = magicZip(monoA, fluxB); // { (a:b1), (a:b2), ... (a:b100), ...}

Каков правильный (или стандартный) способ написания функции magicZip

Answer 1

Вы можете создать этот метод:

private <T>Flux<Tuple2<T, T>> magicZip(Mono<T> mono, Flux<T> flux) {
    Flux<T> repeatableMono = mono.repeat();
    return flux.zipWith(repeatableMono);
}

Пример для типа String:

    Flux<Tuple2<String, String>> test = magicZip(getMono(), getFlux()).doOnNext(objects -> System.out.println(objects.getT1() + objects.getT2()));
    test.blockLast();

Answer 2

Я думаю, что с функцией zip невозможно, потому что она производит столько элементов, сколько наименьшее из двух.

Я думаю, что вы можете достичь этого:

Flux<A> fluxA = monoA.flux();
Flux<Tuple2<A,B>> zippedTuples =fluxB.flatMap(b -> fluxA.map(a -> Tuples.of(a,b)));