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

Replay terminal signals to late subscribers in Flux.replay(int) and Flux.cache(int) #3200

Merged
merged 7 commits into from
Oct 3, 2022
Merged

Replay terminal signals to late subscribers in Flux.replay(int) and Flux.cache(int) #3200

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
946f8c8
Validating replayZero case using FluxPublish modification
chemicL Sep 22, 2022
c23b213
Supporting replay of 0 elements but with terminal signals
chemicL Sep 23, 2022
79313a9
Flux.replay with TTL and 0 history reconnects
chemicL Sep 27, 2022
1dd1390
Re-connects not supported javadoc note
chemicL Sep 27, 2022
ad85fc8
Minor formatting
chemicL Sep 27, 2022
5f71489
Removed unnecessary defensive drain
chemicL Sep 27, 2022
e0eb173
Restored assigning parent to inner to after cancellation
chemicL Sep 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 8 additions & 7 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 @@ -7219,7 +7219,7 @@ public final ConnectableFlux<T> publish() {
*/
public final ConnectableFlux<T> publish(int prefetch) {
return onAssembly(new FluxPublish<>(this, prefetch, Queues
.get(prefetch)));
.get(prefetch), true));
}

/**
Expand Down Expand Up @@ -7565,7 +7565,7 @@ public final ConnectableFlux<T> replay() {
* Will retain up to the given history size onNext signals. Completion and Error will also be
* replayed.
* <p>
* Note that {@code cache(0)} will only cache the terminal signal without
* Note that {@code replay(0)} will only cache the terminal signal without
* expiration.
*
* <p>
Expand All @@ -7579,8 +7579,8 @@ 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 FluxPublish<>(this, Queues.SMALL_BUFFER_SIZE,
Queues.get(Queues.SMALL_BUFFER_SIZE), false));
}
return onAssembly(new FluxReplay<>(this, history, 0L, null));
}
Expand Down Expand Up @@ -7662,8 +7662,8 @@ 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 FluxPublish<>(this, Queues.SMALL_BUFFER_SIZE,
Queues.get(Queues.SMALL_BUFFER_SIZE), false));
}
return onAssembly(new FluxReplay<>(this, history, ttl.toNanos(), timer));
}
Expand Down Expand Up @@ -7987,7 +7987,8 @@ public final <A> Flux<A> scanWith(Supplier<A> initial, BiFunction<A, ? super T,
*/
public final Flux<T> share() {
return onAssembly(new FluxRefCount<>(
new FluxPublish<>(this, Queues.SMALL_BUFFER_SIZE, Queues.small()), 1)
new FluxPublish<>(this, Queues.SMALL_BUFFER_SIZE, Queues.small(), true)
, 1)
);
}

Expand Down
38 changes: 29 additions & 9 deletions reactor-core/src/main/java/reactor/core/publisher/FluxPublish.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
* Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -57,6 +57,11 @@ final class FluxPublish<T> extends ConnectableFlux<T> implements Scannable {

final Supplier<? extends Queue<T>> queueSupplier;

/**
* Whether to prepare for a reconnect after the source terminates.
*/
final boolean resetUponSourceTermination;

volatile PublishSubscriber<T> connection;
@SuppressWarnings("rawtypes")
static final AtomicReferenceFieldUpdater<FluxPublish, PublishSubscriber> CONNECTION =
Expand All @@ -66,13 +71,15 @@ final class FluxPublish<T> extends ConnectableFlux<T> implements Scannable {

FluxPublish(Flux<? extends T> source,
int prefetch,
Supplier<? extends Queue<T>> queueSupplier) {
Supplier<? extends Queue<T>> queueSupplier,
boolean resetUponSourceTermination) {
if (prefetch <= 0) {
throw new IllegalArgumentException("bufferSize > 0 required but it was " + prefetch);
}
this.source = Objects.requireNonNull(source, "source");
this.prefetch = prefetch;
this.queueSupplier = Objects.requireNonNull(queueSupplier, "queueSupplier");
this.resetUponSourceTermination = resetUponSourceTermination;
}

@Override
Expand Down Expand Up @@ -111,7 +118,7 @@ public void subscribe(CoreSubscriber<? super T> actual) {
}

PublishSubscriber<T> c = connection;
if (c == null || c.isTerminated()) {
if (c == null || (this.resetUponSourceTermination && c.isTerminated())) {
PublishSubscriber<T> u = new PublishSubscriber<>(prefetch, this);
if (!CONNECTION.compareAndSet(this, c, u)) {
continue;
Expand All @@ -120,15 +127,24 @@ public void subscribe(CoreSubscriber<? super T> actual) {
c = u;
}

inner.parent = c;
Copy link
Contributor

@OlegDokuka OlegDokuka Sep 29, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks like a behavior change. not sure about the impact but let's keep the old logic as it was

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right. In the earlier stages this was necessary, but it should be only necessary in the previous logic. Restored.


if (c.add(inner)) {
if (inner.isCancelled()) {
c.remove(inner);
}
else {
inner.parent = c;
}
c.drain();
break;
} else if (!this.resetUponSourceTermination) {
if (c.error != null) {
inner.actual.onError(c.error);
} else {
if (c.queue.isEmpty()) {
c.drain();
}
inner.actual.onComplete();
}
break;
}
}
}
Expand Down Expand Up @@ -515,16 +531,20 @@ boolean checkTerminated(boolean d, boolean empty) {
if (d) {
Throwable e = error;
if (e != null && e != Exceptions.TERMINATED) {
CONNECTION.compareAndSet(parent, this, null);
e = Exceptions.terminate(ERROR, this);
if (parent.resetUponSourceTermination) {
CONNECTION.compareAndSet(parent, this, null);
e = Exceptions.terminate(ERROR, this);
}
queue.clear();
for (PubSubInner<T> inner : terminate()) {
inner.actual.onError(e);
}
return true;
}
else if (empty) {
CONNECTION.compareAndSet(parent, this, null);
if (parent.resetUponSourceTermination) {
CONNECTION.compareAndSet(parent, this, null);
}
for (PubSubInner<T> inner : terminate()) {
inner.actual.onComplete();
}
Expand Down
14 changes: 14 additions & 0 deletions reactor-core/src/main/java/reactor/core/publisher/FluxReplay.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1161,6 +1161,20 @@ public void subscribe(CoreSubscriber<? super T> actual) {
}
}

public static void main(String[] args) throws Exception {
ConnectableFlux<Integer> f = Flux.just(1, 2, 3).replay(0);

f.subscribe(System.out::println, System.out::println, () -> System.out.println(
"complete"));
f.connect();

f.subscribe(System.out::println, System.out::println, () -> System.out.println(
"complete"));

Thread.sleep(100);
// f.connect();
}

@Override
public final CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual)
throws Throwable {
Expand Down
75 changes: 74 additions & 1 deletion reactor-core/src/test/java/reactor/core/publisher/FluxCacheTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) 2017-2021 VMware Inc. or its affiliates, All Rights Reserved.
* Copyright (c) 2017-2022 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -125,6 +125,47 @@ public void cacheFluxTTL2() {
.verifyComplete();
}

@Test
void cacheZeroFlux() {
VirtualTimeScheduler vts = VirtualTimeScheduler.create();

Flux<Tuple2<Long, Integer>> source = Flux.just(1, 2, 3)
.delayElements(Duration.ofMillis(1000)
, vts)
.cache(0)
.elapsed(vts);

StepVerifier.withVirtualTime(() -> source, () -> vts, Long.MAX_VALUE)
.thenAwait(Duration.ofSeconds(3))
.expectNextMatches(t -> t.getT1() == 1000 && t.getT2() == 1)
.expectNextMatches(t -> t.getT1() == 1000 && t.getT2() == 2)
.expectNextMatches(t -> t.getT1() == 1000 && t.getT2() == 3)
.verifyComplete();

StepVerifier.create(source).verifyComplete();
}

@Test
public void cacheZeroFluxTTL() {
VirtualTimeScheduler vts = VirtualTimeScheduler.create();

Flux<Tuple2<Long, Integer>> source = Flux.just(1, 2, 3)
.delayElements(
Duration.ofMillis(1000), vts
)
.cache(0, Duration.ofMillis(2000), vts)
.elapsed(vts);

StepVerifier.withVirtualTime(() -> source, () -> vts, Long.MAX_VALUE)
.thenAwait(Duration.ofSeconds(3))
.expectNextMatches(t -> t.getT1() == 1000 && t.getT2() == 1)
.expectNextMatches(t -> t.getT1() == 1000 && t.getT2() == 2)
.expectNextMatches(t -> t.getT1() == 1000 && t.getT2() == 3)
.verifyComplete();

StepVerifier.create(source).verifyComplete();
}

@Test
public void cacheContextHistory() {
AtomicInteger contextFillCount = new AtomicInteger();
Expand Down Expand Up @@ -156,6 +197,38 @@ public void cacheContextHistory() {
assertThat(contextFillCount).as("cacheHit3").hasValue(4);
}

@Test
public void cacheZeroContext() {
AtomicInteger contextFillCount = new AtomicInteger();
Flux<String> cached = Flux.just(1, 2)
.flatMap(i -> Mono.deferContextual(Mono::just)
.map(ctx -> ctx.getOrDefault("a", "BAD"))
)
.cache(0)
.contextWrite(ctx -> ctx.put("a", "GOOD" + contextFillCount.incrementAndGet()));

// at first pass, the Context is propagated to subscriber, but not cached
String cacheMiss = cached.blockLast();
assertThat(cacheMiss).as("cacheMiss").isEqualTo("GOOD1");
assertThat(contextFillCount).as("cacheMiss").hasValue(1);

// at second subscribe, the Context fill attempt is still done, but ultimately
// ignored since source terminated
String zeroCache = cached.blockLast();
assertThat(zeroCache).as("zeroCache").isNull(); //value from the cache
assertThat(contextFillCount).as("zeroCache").hasValue(2); //function was still invoked

//at third subscribe, function is called for the 3rd time, but the context is still cached
String zeroCache2 = cached.blockLast();
assertThat(zeroCache2).as("zeroCache2").isNull();
assertThat(contextFillCount).as("zeroCache2").hasValue(3);

//at fourth subscribe, function is called for the 4th time, but the context is still cached
String zeroCache3 = cached.blockLast();
assertThat(zeroCache3).as("zeroCache3").isNull();
assertThat(contextFillCount).as("zeroCache3").hasValue(4);
}

@Test
public void cacheContextTime() {
AtomicInteger contextFillCount = new AtomicInteger();
Expand Down
Loading