fix: rewrite of the concatenation operators

This fixes race conditions in concatMap and stream concatenation operators. Refs: #1388
smallrye · Nov 30, 2023 · b552fc4 · b552fc4
1 parent f6be4a8
commit b552fc4
Show file tree

Hide file tree

Showing 5 changed files with 217 additions and 354 deletions.
diff --git a/implementation/revapi.json b/implementation/revapi.json
@@ -51,7 +51,14 @@
     "criticality" : "highlight",
     "minSeverity" : "POTENTIALLY_BREAKING",
     "minCriticality" : "documented",
-    "differences" : [ ]
+    "differences" : [
+      {
+        "ignore": true,
+        "code": "java.class.removed",
+        "old": "class io.smallrye.mutiny.operators.multi.MultiConcatMapOp.ConcatMapMainSubscriber<I, O>",
+        "justification": "Internal API refactoring"
+      }
+    ]
   }
 }, {
   "extension" : "revapi.reporter.json",

diff --git a/implementation/src/main/java/io/smallrye/mutiny/helpers/Subscriptions.java b/implementation/src/main/java/io/smallrye/mutiny/helpers/Subscriptions.java
@@ -6,6 +6,7 @@
 import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.smallrye.mutiny.CompositeException;
@@ -103,6 +104,19 @@ public static long add(AtomicLong requested, long requests) {
         }
     }
 
+    public static <T> long add(AtomicLongFieldUpdater<T> updater, T receiver, long requests) {
+        for (;;) {
+            long r = updater.get(receiver);
+            if (r == Long.MAX_VALUE) {
+                return Long.MAX_VALUE;
+            }
+            long u = add(r, requests);
+            if (updater.compareAndSet(receiver, r, u)) {
+                return r;
+            }
+        }
+    }
+
     /**
      * Atomically subtract the given number (positive, not validated) from the target field unless it contains Long.MAX_VALUE.
      *