Merge pull request #378 from smallrye/fix/onCancellationPropagation

Do not trigger onCancellation() when the stream has already completed
smallrye · Dec 3, 2020 · 3ef50de · 3ef50de
2 parents 16a831a + 9fddfee
commit 3ef50de
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 0 deletions.
diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiOnCancellationCall.java b/implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiOnCancellationCall.java
@@ -31,6 +31,12 @@ public MultiOnCancellationCallProcessor(MultiSubscriber<? super T> downstream) {
             super(downstream);
         }
 
+        @Override
+        public void onCompletion() {
+            supplierInvoked.set(true);
+            super.onCompletion();
+        }
+
         @Override
         public void cancel() {
             execute().subscribe().with(

diff --git a/...mentation/src/main/java/io/smallrye/mutiny/operators/multi/MultiOnCancellationInvoke.java b/...mentation/src/main/java/io/smallrye/mutiny/operators/multi/MultiOnCancellationInvoke.java
@@ -30,6 +30,12 @@ public MultiOnCancellationInvokeProcessor(MultiSubscriber<? super T> downstream)
             super(downstream);
         }
 
+        @Override
+        public void onCompletion() {
+            actionInvoked.set(true);
+            super.onCompletion();
+        }
+
         @Override
         public void cancel() {
             if (actionInvoked.compareAndSet(false, true)) {

diff --git a/implementation/src/test/java/io/smallrye/mutiny/operators/MultiOnCancellationCallTest.java b/implementation/src/test/java/io/smallrye/mutiny/operators/MultiOnCancellationCallTest.java
@@ -139,4 +139,36 @@ public void testCancellationAfterOneItem() {
         subscriber.cancel();
         assertThat(counter.get()).isEqualTo(1);
     }
+
+    @Test
+    public void testCancellationPropagatedBeforeCompletion() {
+        AtomicBoolean invokeCalled = new AtomicBoolean();
+        AtomicBoolean callCalled = new AtomicBoolean();
+
+        Multi<Integer> multi = Multi.createFrom().items(1, 2, 3)
+                .onCancellation().invoke(() -> invokeCalled.set(true))
+                .onCancellation().invoke(() -> callCalled.set(true));
+
+        AssertSubscriber<Integer> subscriber = multi.subscribe().withSubscriber(AssertSubscriber.create());
+        subscriber.cancel();
+        subscriber.assertNotTerminated();
+        assertThat(invokeCalled.get()).isTrue();
+        assertThat(callCalled.get()).isTrue();
+    }
+
+    @Test
+    public void testCancellationNotPropagatedAfterCompletion() {
+        AtomicBoolean invokeCalled = new AtomicBoolean();
+        AtomicBoolean callCalled = new AtomicBoolean();
+
+        Multi<Integer> multi = Multi.createFrom().items(1, 2, 3)
+                .onCancellation().invoke(() -> invokeCalled.set(true))
+                .onCancellation().invoke(() -> callCalled.set(true));
+
+        AssertSubscriber<Integer> subscriber = multi.subscribe().withSubscriber(AssertSubscriber.create(69L));
+        subscriber.assertCompleted().assertItems(1, 2, 3);
+        subscriber.cancel();
+        assertThat(invokeCalled.get()).isFalse();
+        assertThat(callCalled.get()).isFalse();
+    }
 }