Completion event fires before subscription event for empty Multi

[julianhowarth]

I'm slightly confused that most of the time, the completion event on a Multi fires after the subscription event has handled any callback. However, when there are no items, and the subscription callback is waiting on something, then the completion event gets fired before the subscription event callback is finished. This can be demonstrated just by switching the subscription event thread.

    Multi.createFrom()
        .items(() -> Stream.of(0))
        .onSubscription()
        .call(
            () ->                
                Uni.createFrom()
                    .<Void>item(
                        () -> {
                          System.out.println("Got subscription");
                          return null;
                        })
                    .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()))
        .onCompletion()
        .invoke(() -> System.out.println("Got completion"))
        .subscribe()
        .with(item -> System.out.println("Got: " + item));

produces expected output:

Got subscription
Got: 0
Got completion

But switching to an empty Stream:

    Multi.createFrom()
        .items(() -> Stream.of())
        .onSubscription()
        .call(
            () ->
                Uni.createFrom()
                    .<Void>item(
                        () -> {
                          System.out.println("Got subscription");
                          return null;
                        })
                    .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()))
        .onCompletion()
        .invoke(() -> System.out.println("Got completion"))
        .subscribe()
        .with(item -> System.out.println("Got: " + item));

gives:

Got completion
Got subscription

I was expecting to see the same ordering in both cases i.e. that the completion event will not be fired until the subscription callback is complete. If this is the intended behaviour then I think it would be helpful for it to be called out prominently in the documentation.

On a related note, I think it would also be helpful to call out that the subscription callback event is called after the supplier of items to the Multi i.e. that although evaluation of the supplier is deferred until subscription, it is not deferred until after the return of the onSubscription event.

Above is running on mutiny 2.6.2 with Java 21.0.4

[jponge]

I need to check this 😃

[jponge]

So I've checked the implementation and StreamBasedMulti does send onSubscribe() before onComplete(), so the protocol is correct.

What you are observing is a race condition due to the stream sending a completion signal immediately, while the onSubscription().call(...) callback is offset to another thread pool, and since call is meant to be asynchronous it may yield this output (it's another concurrent subscription that doesn't affect the main one)

[jponge]

@julianhowarth What's your concrete use case? (just to make sure we're not missing something in the onSubscription().call() operator)

[julianhowarth]

I actually don't have a concrete use case, I came across this by accident in testing something else but it kept failing due to my assumption about the ordering. So, as I said in the original message, if intended I just think the documentation should make it clearer.

It's due to the fact that for non-empty streams you get very used to being able to rely on:

• do something due to subscription starting
• do something with each item
• do something when completed

happening in strict order as completion must come after the items.

[julianhowarth]

I suppose the other situation where it might be surprising is in the case where the callback threw an exception/returned a failed Uni. So where normally you'd expect to complete with a failure, instead you'll get a successful completion. (and lose the fact that the failure occurred?)

[jponge]

I've opened #1678 because there might be a bug here after all, this needs investigations.