Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Rework FluxReplay to avoid hanging, but reject 0 size #2741

Merged
merged 9 commits into from
Sep 9, 2021
Merged

Rework FluxReplay to avoid hanging, but reject 0 size #2741

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c6f78a7
reworks FluxReplay to follow prefetch strategy
Jul 8, 2021
1173592
polishes
Jul 30, 2021
11c8c74
polishes
Jul 30, 2021
d21033c
improves tests which demonstrates prefetch
Jul 30, 2021
565c782
improves tests which demonstrates prefetch
Jul 30, 2021
a90aba2
improves methods docs
Jul 30, 2021
94aff49
Add a warning log and replace replay(0) with publish()
simonbasle Sep 9, 2021
b14391d
revert logging, have Sinks reject historySize < 1
simonbasle Sep 9, 2021
332df64
fix another IllegalArgumentException message
simonbasle Sep 9, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
8 changes: 8 additions & 0 deletions reactor-core/src/main/java/reactor/core/publisher/Flux.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7470,6 +7470,10 @@ public final ConnectableFlux<T> replay() {
*
*/
public final ConnectableFlux<T> replay(int history) {
if (history == 0) {
//TODO Flux.replay with history == 0 doesn't make much sense. This was replaced by Flux.publish, but such calls will be rejected in a future version
return onAssembly(new FluxPublish<>(this, Queues.SMALL_BUFFER_SIZE, Queues.get(Queues.SMALL_BUFFER_SIZE)));
}
return onAssembly(new FluxReplay<>(this, history, 0L, null));
}

Expand Down Expand Up @@ -7549,6 +7553,10 @@ public final ConnectableFlux<T> replay(Duration ttl, Scheduler timer) {
*/
public final ConnectableFlux<T> replay(int history, Duration ttl, Scheduler timer) {
Objects.requireNonNull(timer, "timer");
if (history == 0) {
//TODO Flux.replay with history == 0 doesn't make much sense. This was replaced by Flux.publish, but such calls will be rejected in a future version
return onAssembly(new FluxPublish<>(this, Queues.SMALL_BUFFER_SIZE, Queues.get(Queues.SMALL_BUFFER_SIZE)));
}
return onAssembly(new FluxReplay<>(this, history, ttl.toNanos(), timer));
}

Expand Down