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

fixes breaking change for fromFuture source cancellation #3252

Merged
merged 3 commits into from
Oct 28, 2022
Merged

fixes breaking change for fromFuture source cancellation #3252

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e6d081b
fixes breaking change for `fromFuture` source cancellation
Oct 26, 2022
436f61f
adds overload to suppress cancellation; updates docs
Oct 27, 2022
2b24aa7
improves docs
Oct 27, 2022
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
49 changes: 44 additions & 5 deletions reactor-core/src/main/java/reactor/core/publisher/Mono.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -617,8 +617,8 @@ public static <I> Mono<I> fromDirect(Publisher<? extends I> source){
* <img class="marble" src="doc-files/marbles/fromFuture.svg" alt="">
* <p>
* Note that the future is not cancelled when that Mono is cancelled, but that behavior
* can be obtained by using a {@link #doFinally(Consumer)} that checks
* for a {@link SignalType#CANCEL} and calls {@link CompletableFuture#cancel(boolean)}.
* can be obtained by using a {@link #fromFuture(CompletableFuture, boolean)}
* overload with the {@code cancel} parameter set to {@code true}.
*
* @param future {@link CompletableFuture} that will produce a value (or a null to
* complete immediately)
Expand All @@ -627,7 +627,26 @@ public static <I> Mono<I> fromDirect(Publisher<? extends I> source){
* @see #fromCompletionStage(CompletionStage) fromCompletionStage for a generalization
*/
public static <T> Mono<T> fromFuture(CompletableFuture<? extends T> future) {
return onAssembly(new MonoCompletionStage<>(future));
return fromFuture(future, false);
}

/**
* Create a {@link Mono}, producing its value using the provided {@link CompletableFuture}.
*
* <p>
* <img class="marble" src="doc-files/marbles/fromFuture.svg" alt="">
* <p>
*
* @param future {@link CompletableFuture} that will produce a value (or a null to
* complete immediately)
* @param cancel specifies whether future should be cancelled or not via the
* {@link CompletableFuture#cancel(boolean)} call
* @param <T> type of the expected value
* @return A {@link Mono}.
* @see #fromCompletionStage(CompletionStage) fromCompletionStage for a generalization
*/
public static <T> Mono<T> fromFuture(CompletableFuture<? extends T> future, boolean cancel) {
return onAssembly(new MonoCompletionStage<>(future, cancel));
}

/**
Expand All @@ -638,8 +657,8 @@ public static <T> Mono<T> fromFuture(CompletableFuture<? extends T> future) {
* <img class="marble" src="doc-files/marbles/fromFutureSupplier.svg" alt="">
* <p>
* Note that the future is not cancelled when that Mono is cancelled, but that behavior
* can be obtained by using a {@link #doFinally(Consumer)} that checks
* for a {@link SignalType#CANCEL} and calls {@link CompletableFuture#cancel(boolean)}.
* can be obtained by using a {@link #fromFuture(Supplier, boolean)}
* overload with the {@code cancel} parameter set to {@code true}.
*
* @param futureSupplier The {@link Supplier} of a {@link CompletableFuture} that will produce a value (or a null to
* complete immediately). This allows lazy triggering of future-based APIs.
Expand All @@ -651,6 +670,26 @@ public static <T> Mono<T> fromFuture(Supplier<? extends CompletableFuture<? exte
return defer(() -> onAssembly(new MonoCompletionStage<>(futureSupplier.get())));
}

/**
* Create a {@link Mono} that wraps a {@link CompletableFuture} on subscription,
* emitting the value produced by the Future.
*
* <p>
* <img class="marble" src="doc-files/marbles/fromFutureSupplier.svg" alt="">
* <p>
*
* @param futureSupplier The {@link Supplier} of a {@link CompletableFuture} that will produce a value (or a null to
* complete immediately). This allows lazy triggering of future-based APIs.
* @param cancel specifies whether future should be cancelled or not via the
* {@link CompletableFuture#cancel(boolean)} call
* @param <T> type of the expected value
* @return A {@link Mono}.
* @see #fromCompletionStage(Supplier) fromCompletionStage for a generalization
*/
public static <T> Mono<T> fromFuture(Supplier<? extends CompletableFuture<? extends T>> futureSupplier, boolean cancel) {
return defer(() -> onAssembly(new MonoCompletionStage<>(futureSupplier.get(), cancel)));
}

/**
* Create a {@link Mono} that completes empty once the provided {@link Runnable} has
* been executed.
Expand Down
10 changes: 9 additions & 1 deletion reactor-core/src/main/java/reactor/core/publisher/MonoCompletionStage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import java.util.Objects;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Future;
Expand All @@ -40,9 +41,16 @@ final class MonoCompletionStage<T> extends Mono<T>
implements Fuseable, Scannable {

final CompletionStage<? extends T> future;
final boolean cancel;

MonoCompletionStage(CompletionStage<? extends T> future) {
this.future = Objects.requireNonNull(future, "future");
this.cancel = false;
}

MonoCompletionStage(CompletableFuture<? extends T> future, boolean cancel) {
this.future = Objects.requireNonNull(future, "future");
this.cancel = cancel;
}

@Override
Expand All @@ -52,7 +60,7 @@ public void subscribe(CoreSubscriber<? super T> actual) {
@Override
public void cancel() {
super.cancel();
if (future instanceof Future) {
if (cancel) {
Copy link
Member

Choose a reason for hiding this comment

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

I believe the instanceof check is still required.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

it is impossible that cancel == true and !future instanceof Future. The public API allows only CompletableFuture to have the cancel flag configurable. in all other cases, it is false. This, checking instance of in addition to cancel is true will be overhead

Copy link
Contributor Author

Choose a reason for hiding this comment

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

check the following code -> https://github.com/reactor/reactor-core/pull/3252/files#diff-e815d2e7a55f9f17006048ff085897e713143567345688c33e9d64b19f911f06R51

((Future<?>) future).cancel(true);
}
}
Expand Down
25 changes: 23 additions & 2 deletions reactor-core/src/test/java/reactor/core/publisher/MonoCompletionStageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ public void propagateCancellationToCompletionFuture() {
CompletableFuture<Integer> future = new CompletableFuture<>();

Mono<Integer> mono = Mono
.fromFuture(future);
.fromFuture(future, true);

StepVerifier.create(mono)
.expectSubscription()
Expand All @@ -50,14 +50,35 @@ public void propagateCancellationToCompletionFuture() {
}

@Test
public void cancelThenFutureFails() {
public void cancelThenFutureFailsWithDroppedError() {
CompletableFuture<Integer> future = new CompletableFuture<>();
AtomicReference<Subscription> subRef = new AtomicReference<>();

Mono<Integer> mono = Mono
.fromFuture(future)
.doOnSubscribe(subRef::set);

StepVerifier.create(mono)
.expectSubscription()
.then(() -> {
subRef.get().cancel();
future.completeExceptionally(new IllegalStateException("boom"));
future.complete(1);
})
.thenCancel()//already cancelled but need to get to verification
.verifyThenAssertThat()
.hasDroppedErrorWithMessage("boom");
}

@Test
public void cancelThenFutureFails() {
CompletableFuture<Integer> future = new CompletableFuture<>();
AtomicReference<Subscription> subRef = new AtomicReference<>();

Mono<Integer> mono = Mono
.fromFuture(future, true)
.doOnSubscribe(subRef::set);

StepVerifier.create(mono)
.expectSubscription()
.then(() -> {
Expand Down