reactor zip two sources with conditions

I have two sources of sorted integers: A and B.
And I need to zip the two sources into one the restriction: Only zip these items who exist both in A and B.

For example:
A is: 1 2 3 10 11 12 13 14
B is: 10 12 13 14 15 16
The reactor's zip operator would generate (1,10),(2,12),(3,13),(10,14),(11,15),(12,16)
But I want to get this source: (10,10),(12,12),(13,13),(14,14)

I tried a zip and match approach below, but failed, because the first complete source will cause zip operator too complete the result source, and thus left source integer unmatched.

public static Flux<Tuple2<Integer, Integer>> match(Flux<Tuple2<Integer, Integer>> flux) {
        Tuple2<Integer, Integer> invalidValue = Tuples.of(0, 0);
        return flux.compose(f -> f.map(new Function<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
            private final Queue<Integer> left = new ArrayDeque<>(); // here is a problem: when complete, left or right queue may still have some data.
            private final Queue<Integer> right = new ArrayDeque<>();

            @Override
            public Tuple2<Integer, Integer> apply(Tuple2<Integer, Integer> pair) {
                left.add(pair.getT1());
                right.add(pair.getT2());

                return tryMatch();
            }

            private Tuple2<Integer, Integer> tryMatch() {
                if (left.isEmpty() || right.isEmpty()) {
                    return invalidValue;
                }
                Integer t1 = left.peek();
                Integer t2 = right.peek();
                if (t1.equals(t2)) {
                    left.poll();
                    right.poll();
                    return Tuples.of(t1, t2);
                } else if (t1 < t2) {
                    Integer item = left.poll();
                    log.warn("not matched: {}", item);
                    return tryMatch();
                } else {
                    Integer item = right.poll();
                    log.warn("not matched: {}", item);
                    return tryMatch();
                }
            }
        })).filter(pair -> pair != invalidValue);
    }

Could someone help me, tell me what operator can I use?

No need to map it yourself you can use join and filter

Flux<Integer> f1 = Flux.just(1,2,3,10,11,12,13,14);
Flux<Integer> f2 = Flux.just(10,12,13,14,15,16);

f1.join(f2,f ->Flux.never(),f-> Flux.never(),Tuples::of)
                .filter(t -> t.getT1().equals(t.getT2()));

Or if you want a function

public <T> Flux<Tuple2<T,T>> match(Flux<T> f1, Flux<T> f2){

    return f1.join(f2,f ->Flux.never(),f-> Flux.never(),Tuples::of)
            .filter(t -> t.getT1().equals(t.getT2()));

}