FluxReceive hangs, sometimes, on subsequent subscribers

[rstoyanchev]

FluxReceive rejects multiple subscribers but only if the second subscribes while first one is still active. Once a subscriber completes and the receiver field is cleared, a second subscriber will not be rejected and will not receive further signals.

One reason to read a second time is to ensure a client response has been drained from at a place where it isn't known if it has been consumed or not. The second attempt to read relies on either draining successfully or getting an error, not hanging indefinitely.

[rstoyanchev]

The following workaround was added to 5.1.3, which led to #540 and SPR-17564. It turns out the root cause was that both Reactor Netty and the Spring Framework were signalling IIlegalStateException on a subsequent subscriber, if the response was read in full by the first subscriber, but it works okay if the first subscriber cancels (e.g. response.bodyToMono(Void.class)), because in that case only the Spring Framework signals an error on subsequent subscribers. This was quite difficult to find out by the way. Here is the current behavior targetting 5.1.4.

The current behavior seems broken and our workaround proved fragile. I see no good reasons why subsequent subscribers, after a cancellation, should not be rejected as is the case with multiple subscribers during the course of reading. Even if there is some intent like retries, certainly indefinite hanging doesn't seem right either.

Once again the reason for talking about multiple subscribers in the first place is the need to ensure that response is drained always, and sometimes we can't be sure if it has been consumed or not. So we try to consume again and rely on being kicked out if it there has been a consumer already. If you'd like to keep the current behavior, then please consider exposing a way to do the equivalent of response.bodyToMono(Void.class). That's used where no response body is expected to ensure the body is drained, by attempting to read and cancel if any data shows up.

[rstoyanchev]

@violetagg the following test here can be used to debug the issue. The scenario is this:

1. An error response that is consumed with an onStatus handler which uses bodyToMono(Void.class) which does takeWhile and cancels if it finds any data.
2. Cancellation leads to HttpClientOperations#onInboundCancel which closes the channel.
3. A second subscription then from wrapper around the onStatus handler tries to consume again and ignore errors. That one goes to FluxReceiver#startReceiver but since at this point inboundDone is false it simply takes the subscription and nothing ever happens again.

I believe this behavior is somewhat intentional to keep the door open to allow re-subscribing but I'm not sure there is a clear purpose at this time. @smaldini any more insight?

[violetagg]

@rstoyanchev This PR #1185 should fix the issue. Can you please verify it?

[violetagg]

I'm closing this issue as a fix is provided with #1185