Merge pull request #471 from smallrye/bug/uni-cancelled-future

smallrye · Feb 10, 2021 · 7256672 · 7256672
2 parents 1e49045 + 741904c
commit 7256672
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 5 deletions.
diff --git a/...entation/src/main/java/io/smallrye/mutiny/operators/uni/builders/UniCreateFromFuture.java b/...entation/src/main/java/io/smallrye/mutiny/operators/uni/builders/UniCreateFromFuture.java
@@ -4,6 +4,7 @@
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
 import io.smallrye.mutiny.infrastructure.Infrastructure;
@@ -66,16 +67,26 @@ private void dispatchImmediateResult(Future<? extends T> future, UniSubscriber<?
     }
 
     private void dispatchDeferredResult(Future<? extends T> future, UniSubscriber<? super T> downstream) {
-        downstream.onSubscribe(() -> future.cancel(false));
+        AtomicBoolean cancelled = new AtomicBoolean(false);
+        downstream.onSubscribe(() -> {
+            cancelled.set(true);
+            future.cancel(false);
+        });
         // Because future.get is blocking, we must use a separated thread.
         Infrastructure.getDefaultExecutor().execute(() -> {
             try {
                 T item = future.get();
-                downstream.onItem(item);
+                if (!cancelled.get()) {
+                    downstream.onItem(item);
+                }
             } catch (ExecutionException e) {
-                downstream.onFailure(e.getCause());
+                if (!cancelled.get()) {
+                    downstream.onFailure(e.getCause());
+                }
             } catch (Exception e) {
-                downstream.onFailure(e);
+                if (!cancelled.get()) {
+                    downstream.onFailure(e);
+                }
             }
         });
     }

diff --git a/implementation/src/test/java/io/smallrye/mutiny/operators/UniCreateFromFutureTest.java b/implementation/src/test/java/io/smallrye/mutiny/operators/UniCreateFromFutureTest.java
@@ -10,6 +10,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 import io.smallrye.mutiny.Uni;
@@ -221,7 +222,7 @@ public void testThatSubscriberCanCancelBeforeEmission() {
         subscriber.assertNotTerminated();
     }
 
-    @Test
+    @RepeatedTest(10)
     public void testThatSubscriberCanCancelBeforeEmissionWithSupplier() {
         UniAssertSubscriber<Integer> subscriber = UniAssertSubscriber.create();
         CompletableFuture<Integer> cs = new CompletableFuture<>();