Merge pull request #1521 from smallrye/fix/1520

Fix cancellation issues in Uni to Multi bridge
smallrye · Feb 21, 2024 · aeb71fa · aeb71fa
2 parents bc3ba4f + 607320e
commit aeb71fa
Show file tree

Hide file tree

Showing 17 changed files with 38 additions and 22 deletions.
diff --git a/bom/pom.xml b/bom/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>io.smallrye.reactive</groupId>
     <artifactId>mutiny-bom</artifactId>
-    <version>2.5.6</version>
+    <version>2.5.7-SNAPSHOT</version>
 
     <packaging>pom</packaging>
 

diff --git a/context-propagation/pom.xml b/context-propagation/pom.xml
@@ -7,7 +7,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-smallrye-context-propagation</artifactId>

diff --git a/documentation/pom.xml b/documentation/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-documentation</artifactId>

diff --git a/implementation/pom.xml b/implementation/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny</artifactId>

diff --git a/implementation/src/main/java/io/smallrye/mutiny/converters/uni/UniToMultiPublisher.java b/implementation/src/main/java/io/smallrye/mutiny/converters/uni/UniToMultiPublisher.java
@@ -58,19 +58,21 @@ public Context context() {
 
         @Override
         public void cancel() {
-            if (upstream != null) {
-                upstream.cancel();
+            if (STATE_UPDATER.getAndSet(this, State.DONE) != State.DONE) {
+                if (upstream != null) {
+                    upstream.cancel();
+                }
             }
         }
 
         @Override
         public void request(long n) {
-            if (n <= 0L) {
-                downstream.onError(new IllegalArgumentException("Invalid request"));
-                return;
-            }
             if (STATE_UPDATER.compareAndSet(this, State.INIT, State.UNI_REQUESTED)) {
-                AbstractUni.subscribe(uni, this);
+                if (n <= 0L) {
+                    onFailure(new IllegalArgumentException("Invalid request"));
+                } else {
+                    AbstractUni.subscribe(uni, this);
+                }
             }
         }
 

diff --git a/implementation/src/test/java/io/smallrye/mutiny/BugReproducersTest.java b/implementation/src/test/java/io/smallrye/mutiny/BugReproducersTest.java
@@ -4,6 +4,7 @@
 import static org.awaitility.Awaitility.await;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -94,4 +95,17 @@ void reproducer_quarkus_21528() {
 
         Uni.join().first(asList).withItem();
     }
+
+    @Test
+    void reproducer_1520() {
+        // From https://github.com/smallrye/smallrye-mutiny/issues/1520 and
+        // to address https://github.com/quarkusio/quarkus/issues/34613
+        AtomicInteger counter = new AtomicInteger();
+        Multi.createFrom().iterable(List.of("aa", "bb", "cc"))
+                .collect().asList()
+                .onTermination().invoke(counter::incrementAndGet)
+                .onFailure().retry().withBackOff(Duration.ofSeconds(1)).atMost(2)
+                .await().atMost(Duration.ofSeconds(5));
+        assertThat(counter).hasValue(1);
+    }
 }
diff --git a/kotlin/pom.xml b/kotlin/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-kotlin</artifactId>

diff --git a/math/pom.xml b/math/pom.xml
@@ -7,7 +7,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <name>SmallRye Mutiny - Math operators</name>

diff --git a/pom.xml b/pom.xml
@@ -11,7 +11,7 @@
 
     <groupId>io.smallrye.reactive</groupId>
     <artifactId>mutiny-project</artifactId>
-    <version>2.5.6</version>
+    <version>2.5.7-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <name>SmallRye Mutiny - Parent</name>

diff --git a/reactive-streams-operators-jakarta/pom.xml b/reactive-streams-operators-jakarta/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-reactive-streams-operators-jakarta</artifactId>

diff --git a/reactive-streams-operators/pom.xml b/reactive-streams-operators/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-reactive-streams-operators</artifactId>

diff --git a/reactive-streams-tck-tests/pom.xml b/reactive-streams-tck-tests/pom.xml
@@ -7,7 +7,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>reactive-streams-tck-tests</artifactId>

diff --git a/reactor/pom.xml b/reactor/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-reactor</artifactId>

diff --git a/release/pom.xml b/release/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>smallrye-mutiny-release</artifactId>

diff --git a/rxjava3/pom.xml b/rxjava3/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-rxjava3</artifactId>

diff --git a/test-utils/pom.xml b/test-utils/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <artifactId>mutiny-test-utils</artifactId>

diff --git a/workshop-examples/pom.xml b/workshop-examples/pom.xml
@@ -7,7 +7,7 @@
     <parent>
         <groupId>io.smallrye.reactive</groupId>
         <artifactId>mutiny-project</artifactId>
-        <version>2.5.6</version>
+        <version>2.5.7-SNAPSHOT</version>
     </parent>
 
     <name>SmallRye Mutiny - Workshop examples</name>