Replay terminal signals to late subscribers in Flux.replay(int) and Flux.cache(int) #3200
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
chemicL
commented
Sep 23, 2022
simonbasle
reviewed
Sep 27, 2022
reactor-core/src/main/java/reactor/core/publisher/FluxReplay.java
Outdated
Show resolved
Hide resolved
simonbasle
reviewed
Sep 27, 2022
reactor-core/src/main/java/reactor/core/publisher/FluxPublish.java
Outdated
Show resolved
Hide resolved
chemicL
changed the title
OlegDokuka
suggested changes
Sep 29, 2022
| @@ -120,15 +127,21 @@ public void subscribe(CoreSubscriber<? super T> actual) { | |||
| c = u; | |||
| } | |||
|
|
|||
| inner.parent = c; | |||
There was a problem hiding this comment.
looks like a behavior change. not sure about the impact but let's keep the old logic as it was
There was a problem hiding this comment.
Right. In the earlier stages this was necessary, but it should be only necessary in the previous logic. Restored.
OlegDokuka
approved these changes
Oct 3, 2022
|
@chemicL this PR seems to have been merged on a maintenance branch, please ensure the change is merge-forwarded to intermediate maintenance branches and up to |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Supporting the caching of only terminal signals in case of
Flux.replay(int)andFlux.cache(int)operators.As it currently stands, these operators, when provided
0as the argument, resort to behaving likeFlux.publish()which differs in the way termination signals are handled by the> 0cases. This change still uses theFluxPublishclass to implement the logic, however with a few minor changes to its implementation.First of all,
FluxPublishresets itself after source termination to be able toconnect()again.FluxReplayon the other hand, does not reset itself in the case of only buffering signals without timeout. Therefore, for the behaviour of caching the terminals,FluxPublishdoes not reset itself and replays the terminal when it receives a late subscription.For cases with expiry (TTL arguments),
FluxReplaydoes indeed reset itself. Therefore, the behaviour for the time constrained values remains as before, usingFluxPublishimplementation for the0history case, but without caching terminals, while not honouring the TTL. This case can be later implemented if needed.