Cancellation of upstreams when the merged downstream terminates

[vincent-louis]

I have a Multi<> stream created by merging several other Multi<> streams :
Multi.createBy().merging().streams(multi1, multi2, muti3)...

I notice that when the merged multi1 stream terminates on failure, then the failure is propagated downstream to the merged stream, which also terminates on failure: this is exactly what expected ✔️

However, I also notice that after the downstream termination, then the other 2 upstreams multi2 and multi3 (which have not failed) are not cancelled by the downstream termination, so that they go on running although the merged downstream is stopped and doesn't consume items any longer. Is it the expected behavior? ❓

If so, what is the right way of making all not failed upstreams to be cancelled?

I updated my code in:

var subscriptions = new Flow.Subscription[] {null, null, null};
Multi.createBy().merging().streams(
    multi1.onSubscription().invoke(it -> subscriptions[0] = it),
    multi2.onSubscription().invoke(it -> subscriptions[1] = it),
    multi3.onSubscription().invoke(it -> subscriptions[2] = it)
)
.log("MERGED STREAM")
.onTermination().invoke(() -> Arrays.stream(subscriptions).forEach(Flow.Subscription::cancel))
...

It works, I can see in the logs that this time, multi2 and multi3 and correctly cancelled.
However I find this code too much verbose...
I would be interested in finding a more concise way of doing this.

Thanks for you help!

[jponge]

I'm looking into this, it could be a bug.

[jponge]

See #1646 thanks @vincent-louis 😃

[vincent-louis]

Thanks a lot Julien for your reactivity!