adds cancellation of the source future support (#3146)

Author: OlegDokuka

reactor-core/src/main/java/reactor/core/publisher/MonoCompletionStage.java

@@ -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.
@@ -20,6 +20,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Future;
 
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
@@ -47,7 +48,15 @@ final class MonoCompletionStage<T> extends Mono<T>
     @Override
     public void subscribe(CoreSubscriber<? super T> actual) {
         Operators.MonoSubscriber<T, T>
-                sds = new Operators.MonoSubscriber<>(actual);
+                sds = new Operators.MonoSubscriber<T, T>(actual) {
+            @Override
+            public void cancel() {
+                super.cancel();
+                if (future instanceof Future) {
+                    ((Future<?>) future).cancel(true);
+                }
+            }
+        };
 
         actual.onSubscribe(sds);
 

reactor-core/src/test/java/reactor/core/publisher/MonoCompletionStageTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2015-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.
@@ -33,6 +33,22 @@
 public class
 MonoCompletionStageTest {
 
+	//https://github.com/reactor/reactor-core/issues/3138
+	@Test
+	public void propagateCancellationToCompletionFuture() {
+		CompletableFuture<Integer> future = new CompletableFuture<>();
+
+		Mono<Integer> mono = Mono
+				.fromFuture(future);
+
+		StepVerifier.create(mono)
+		            .expectSubscription()
+		            .thenCancel()
+		            .verify();
+
+		assertThat(future).isCancelled();
+	}
+
 	@Test
 	public void cancelThenFutureFails() {
 		CompletableFuture<Integer> future = new CompletableFuture<>();
@@ -45,13 +61,13 @@ public void cancelThenFutureFails() {
 		StepVerifier.create(mono)
 		            .expectSubscription()
 		            .then(() -> {
-		            	subRef.get().cancel();
-		            	future.completeExceptionally(new IllegalStateException("boom"));
-		            	future.complete(1);
+			            subRef.get().cancel();
+			            future.completeExceptionally(new IllegalStateException("boom"));
+			            future.complete(1);
 		            })
 		            .thenCancel()//already cancelled but need to get to verification
 		            .verifyThenAssertThat()
-		            .hasDroppedErrorWithMessage("boom");
+		            .hasNotDroppedErrors();
 	}
 
 	@Test