fix: race condition on cancellation in UniCallbackSubscriber

Fixes #1494
smallrye · Jan 23, 2024 · be54f15 · be54f15
1 parent e2bb9d9
commit be54f15
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 1 deletion.
diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/UniCallbackSubscriber.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/UniCallbackSubscriber.java
@@ -79,7 +79,7 @@ public final void onItem(T x) {
     @Override
     public void cancel() {
         UniSubscription sub = SUBSCRIPTION_UPDATER.getAndSet(this, CANCELLED);
-        if (sub != CANCELLED) {
+        if (sub != null && sub != CANCELLED) {
             sub.cancel();
         }
     }

diff --git a/implementation/src/test/java/io/smallrye/mutiny/groups/UniSubscriberTest.java b/implementation/src/test/java/io/smallrye/mutiny/groups/UniSubscriberTest.java
@@ -5,6 +5,9 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -15,7 +18,10 @@
 import org.junit.jupiter.api.parallel.ResourceLock;
 
 import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.operators.AbstractUni;
 import io.smallrye.mutiny.subscription.Cancellable;
+import io.smallrye.mutiny.subscription.UniSubscriber;
+import io.smallrye.mutiny.subscription.UniSubscription;
 import junit5.support.InfrastructureResource;
 
 @ResourceLock(value = InfrastructureResource.NAME, mode = ResourceAccessMode.READ)
@@ -105,4 +111,30 @@ public void testTwoCallbacksVariant() {
         cancellable.cancel();
     }
 
+    @Test
+    public void cancellationBeforeSubscriptionArrivesDoesntThrowNpe() {
+        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+        try {
+            AbstractUni<String> uni = new AbstractUni<>() {
+                @Override
+                public void subscribe(UniSubscriber<? super String> subscriber) {
+                    scheduler.schedule(() -> {
+                        subscriber.onSubscribe(new UniSubscription() {
+                            @Override
+                            public void cancel() {
+
+                            }
+                        });
+                        subscriber.onItem("Yolo");
+                    }, 10, TimeUnit.SECONDS);
+                }
+            };
+
+            AtomicReference<String> result = new AtomicReference<>();
+            uni.subscribe().with(result::set).cancel();
+            assertThat(result).hasNullValue();
+        } finally {
+            scheduler.shutdown();
+        }
+    }
 }