Open jGleitz opened 2 years ago
To elaborate on my expectation:
I am using #publish
with a source that honours backpressure. I am relying on the documented property of #publish
that
if any Subscriber is missing demand (requested = 0), multicast will pause pushing/pulling.
I take that to mean that if there are no Subscribers at all (and, hence, no request), pushing and pulling will also pause. This works as expected for next signals, but not for completion. I think it is because of the prefetch: prefetched next signals are buffered and replayed, but completion signals are not.
My expectation is that at least the first subscriber that subscribes after the source completed sees the completion. I would find it even more useful if all later subscribers got the completion replayed, but I guess one could also make the argument that this would be inconsistent with how #publish
handles next signals.
Here is another example that shows why I find the current behaviour unexpected: The completion signal is being buffered while there are no subscribers, as long as there are still outstanding next signals.
We can see this in this succeeding test case:
@Test
void subscribeAfterComplete() {
var sink = Sinks.many().unicast().<Integer>onBackpressureBuffer();
var published = sink.asFlux().log("source").publish().autoConnect().log("published");
sink.tryEmitNext(1).orThrow();
StepVerifier.create(published)
.expectNext(1)
.thenCancel()
.verify(Duration.ofSeconds(1));
sink.tryEmitNext(2).orThrow();
sink.tryEmitComplete().orThrow();
StepVerifier.create(published)
.expectNext(2)
.expectComplete()
.verify(Duration.ofSeconds(1));
}
In other words: #publish
makes sure that no signals are lost while there are no Subscribers, except if the only outstanding signal is a completion. In that case it is dropped.
This indeed sounds like something that needs fixing. Thanks for adding that other test case, we'll look into it 👍
@jGleitz there is a workaround as of now which is replacement of publish
with replay(history=0)
.
In the duplicate issue #3739 the published flux gladly accepts the 2nd late subscriber then ignores it and the pipeline hangs. This behavior makes no sense to me. It is also not documented anywhere, AFAICT.
Stating the obvious that causing a pipeline to hang, possibly in a subtle way, should be avoided.
IMHO, it should simply give it a complete signal, instead of hanging.
Expected Behavior
When a
#publish
ed source completes while no one is subscribed, at least the first Subscriber to subscribe afterwards directly receivesonComplete
.Actual Behavior
The subscriber receives no signal at all and effectively hangs.
Steps to Reproduce
Output:
Workaround
Use
#materialize
before and#dematerialize
after publishing:Your Environment
netty
, ...): nonejava -version
):openjdk version "11.0.11" 2021-04-20
uname -a
):Linux Lovelace 5.13.0-27-generic #29-Ubuntu SMP Wed Jan 12 17:36:47 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux