[Suggestion] Create an operator that merges multiple ordered flux's into a single flow with optional fields for flux's with gaps in there keys

Combine a Flux.zip-akin operator with a key-selecting variant of Flux.mergeComparing for publishers that should be merged based on keys, for both finite and unbounded sources, of any combination in length.

Motivation

I use Reactor everyday in my data pipeline work, to pretty great success. The lazy operators are amazing at handling complex merge operations across many distinct sources. One of the things I run into however is the case when I am trying to fan-in multiple sources of data that have different lengths. and mismatched (but ordered) keys.

Example use-case

An example of this would be merging in 4 different JSON arrays, where a "match-key" would be missing from some of the sets, or that some of the sets have totally different lengths, and would short circuit early.

I have used Flux.groupBy in the past, but that doesn't work in a unbounded Flux case I tend to create a custom interleave for these situations, but a generic solution would be incredibly helpful.

Desired solution

An example signature for this kind of operator that I have experimented with:

    /**
     * This operator merges 4 different flux's together into a single flux based on matching keys.
     * In the case of a source either not having a matched key, or ending early, an empty optional is returned.
     * The Flux's do not have to be the same length, and may have different(but ordered) keys
     * <br>
     * Each source is read until their end.
     * It's assumed that all the sources are already ordered, and that K is comparable
     * @param <K> the key type; Required to be comparable. The smallest value is picked to combine
     * @param <T1> type of the value from source1
     * @param <T2> type of the value from source2
     * @param <T3> type of the value from source3
     * @param <T4> type of the value from source4
     * @param source1 The first Publisher source to combine values from
     * @param source2 The second Publisher source to combine values from
     * @param source3 The third Publisher source to combine values from
     * @param source4 The forth Publisher source to combine values from
     * @param prefetch the minimum size of the internal queue per flux
     * @return a flux based on the produced combinations
     */
    public static <K extends Comparable<? super K>, T1, T2, T3, T4>
    Flux<Tuple5<K, Optional<T1>, Optional<T2>, Optional<T3>, Optional<T4>>>
    zipOnKeyOptional(Flux<? extends Map.Entry<K,T1>> source1,
                     Flux<? extends Map.Entry<K,T2>> source2,
                     Flux<? extends Map.Entry<K,T3>> source3,
                     Flux<? extends Map.Entry<K,T4>> source4, int prefetch);

Desired output

---
title: s
---
stateDiagram-v2    
    sourceOne --> Combiner
    sourceTwo --> Combiner
    sourceThree --> Combiner
    sourceFour --> Combiner


    state sourceOne {
        s11: (1,1)
        s12: (2,2)
        s13: (3,3)
        s14: (4,4)
        s15: (5,5)

        [*] --> s11
        s11 --> s12
        s12 --> s13
        s13 --> s14
        s14 --> s15
        s15 --> [*]
    }

    state sourceTwo {
        [*] --> [*]
    }

    state sourceThree {
        s31: (1,1)
        s32: (2,2)
        s33: (4,4)

        [*] --> s31
        s31 --> s32
        s32 --> s33 
        s33 --> [*]
    }

    state sourceFour {
        s41: (1,1)
        s42: (3,3)
        s43: (4,4)

        [*] --> s41
        s41 --> s42
        s42 --> s43 
        s43 --> [*]
    }


    state Combiner {
        sc1: 1 [1, null, 1,    1]
        sc2: 2 [2, null, 2,    null]  
        sc3: 3 [3, null, null, 3]
        sc4: 4 [4, null, 4,    4]
        sc5: 5 [5, null, null, null]
        [*] --> sc1
        sc1 --> sc2
        sc2 --> sc3
        sc3 --> sc4
        sc4 --> sc5
        sc5 --> [*]
    }

Test Case

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import reactor.util.function.Tuple5;
import reactor.util.function.Tuples;

import java.util.Map;
import java.util.Optional;

import static java.util.Map.entry;
import static java.util.Optional.of;
import static java.util.Optional.empty;

public class ReactiveUtilsTest {
    @Test
    void testZipOnKeyOptional() {
        Flux<Map.Entry<Integer, Integer>> fluxOne = Flux.range(1,5).map(i -> entry(i,i));
        Flux<Map.Entry<Integer, Integer>> fluxTwo = Flux.empty();
        Flux<Map.Entry<Integer, Integer>> fluxThree = Flux.just(entry(1,1), entry(2,2), entry(4,4));
        Flux<Map.Entry<Integer, Integer>> fluxFour = Flux.just(entry(1,1), entry(3,3), entry(4,4));

        Flux<Tuple5<Integer, Optional<Integer>, Optional<Integer>, Optional<Integer>, Optional<Integer>>>
                actual = Flux.zipOnKeyOptional(fluxOne, fluxTwo, fluxThree, fluxFour, 4);
        StepVerifier.create(actual)
                .expectNext(Tuples.of(1, of(1), empty(), of(1), of(1)))
                .expectNext(Tuples.of(2, of(2), empty(), of(2), empty()))
                .expectNext(Tuples.of(3, of(3), empty(), empty(), of(3)))
                .expectNext(Tuples.of(4, of(4), empty(), of(4), of(4)))
                .expectNext(Tuples.of(5, of(5), empty(), empty(), empty()))
                .verifyComplete();
    }
}

Considered alternatives

• Flux.groupBy doesn't work in unbounded / infinite publisher situations.
• groupedFlux's also can't be joined in a structured-concurrency kind of way, like Mono.zip
• I typically implement these functions by having:
  1. Having each flux be mapped to a marker interface that allows me to apply them to a POJO builder
  2. Merging the Flux's that now are cast to the marker interface with an operator like Flux.mergeComparingDelayError(...)
  3. Use Flux.windowUntilChanged to group the entities
  4. flatMap with reduceWith accumulator to build the tuple out
  5. the mapped object is aligned key wise, and has sane default Optional.empty() for unmatched fields