Document Mono.zip subscription scheme when Mono<Void> is supplied

[a701440]

The test below fails when using Mono, if you slightly change the code to use Mono it passes.
3 subscribes are expected, but only one is issued.

    @Test
    public void testZip() {
        AtomicInteger cnt = new AtomicInteger();
        Mono<Void> m1 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Void> m2 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Void> m3 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Void> m4 = m1.zipWith(Mono.zip(List.of(m2,m3), objects -> true), (value1, value2) -> value1);
        m4.block();
        assertEquals(3, cnt.get());
    }

[a701440]

This fails on 3.5.0, but was passing on 3.4.6.

[OlegDokuka]

@a701440 that is expected. MonoZip does not guarantee all sources are subscribed when one of the given completes or errors immediately. What you have seen in 3.4.x line was a design flaw. Consider using Mono.when instead if you need to wait for all sources to be subscribed and completed

[a701440]

It's very confusing that the behavior is very different if you change the code to Integer.

@Test
    public void testZip() {
        AtomicInteger cnt = new AtomicInteger();
        Mono<Integer> m1 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success(1);
        });
        Mono<Integer> m2 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success(2);
        });
        Mono<Integer> m3 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success(3);
        });
        Mono<Integer> m4 = m1.zipWith(Mono.zip(List.of(m2,m3), objects -> true), (value1, value2) -> value1);
        m4.block();
        assertEquals(3, cnt.get());
    }

In my actual case it was used with function.apply. When function was void it did not invoke, when function had a return value it did.

[OlegDokuka]

@a701440 That is expected!

As I stated above, zip is designed to join corresponding values.
By design, it waits for all sources to emit an onNext signal (the last case corresponds to the expectation).

In case one of the sources is empty, zip unsubscribes from the other (if was subscribed before).
To emphasize one more time, what you have seen before was misbehaving.

[OlegDokuka]

what we need to document here is an alternative approach for zipping using toSingleOptional().

[RomaKudryavtsev]

Hi there! I will be glad to investigate and document this issue. Please let me know if it is still relevant.

[OlegDokuka]

@RomaKudryavtsev just do it!

[RomaKudryavtsev]

@OlegDokuka , in javadoc to zip it is stated that: "An error or empty completion of any source will cause other sources to be cancelled and the resulting Mono to immediately error or complete, respectively." I believe this what you are referring to in your responses above.
At the same time, the below test case will be passed:

    @Test
    public void testZipVoid() {
        AtomicInteger cnt = new AtomicInteger();
        Mono<Void> mono1 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Void> mono2 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Void> zippedMono = Mono.zip(mono1, mono2, (v1, v2) -> v1).log();
        zippedMono.block();
        assertEquals(2, cnt.get());
    }

Why would it pass? According to javadoc, I would say that the logic has to be as follows: mono1 completes empty, therefore zippedMono should complete immediately (i.e. without waiting for mono2). But it follows from the test case that it is quite opposite - both mono1 and mono2 are completed. Or maybe it would be proper to say that indeed .zip() simply does not guarantee that all of the publishers will be completed (meaning that in fact they might be completed). Sorry for being boring - I just want to understand the issue properly in order to add this to documentation. Also just want to note that I am using 3.5.0 version

[OlegDokuka]

@RomaKudryavtsev your example is rather coincidence which happens now but in the future it may be different. The correct answer - there is no guarantee

[chemicL]

I updated the title of the issue to reflect the current scope of work.

[RomaKudryavtsev]

@chemicL , hi! I will do it this week - my input will be based on above reply from @OlegDokuka

[RomaKudryavtsev]

@chemicL , @OlegDokuka - please let me know should you have any comments/mark-up.

[chemicL]

Thanks @RomaKudryavtsev and congrats on your first contribution!