simplify end loop

Scottmitch · Nov 15, 2024 · d01f872 · d01f872
1 parent b951e36
commit d01f872
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 16 deletions.
diff --git a/...oncurrent-api/src/main/java/io/servicetalk/concurrent/api/PublisherConcatMapIterable.java b/...oncurrent-api/src/main/java/io/servicetalk/concurrent/api/PublisherConcatMapIterable.java
@@ -246,26 +246,19 @@ private void tryDrainIterator(ErrorHandlingStrategyInDrain errorHandlingStrategy
                                     doCancel();
                                 }
                             } else if (terminalNotification == null && !hasNext) {
-                                for (;;) {
-                                    final Iterator<? extends U> nextIter = iterator;
-                                    if (nextIter == currIter && currRequestN > 0 &&
-                                            (currIter != EmptyIterator.instance() || thrown)) {
+                                final Iterator<? extends U> nextIter = iterator;
+                                if (nextIter == currIter && currRequestN > 0 &&
+                                        (currIter != EmptyIterator.instance() || thrown) &&
                                         // We only request 1 at a time, and therefore we don't have outstanding demand.
                                         // We will not be getting an onNext call, so we write the currIter variable
                                         // before we unlock emitting so visibility to other threads is taken care of
                                         // by the write to emitting below (and later read).
-                                        if (iterUpdater.compareAndSet(this, currIter, EmptyIterator.instance())) {
-                                            if (sourceSubscription != null) {
-                                                sourceSubscription.request(1);
-                                            }
-                                            break;
-                                        }
-                                    } else {
-                                        // if nextIter != currIter -> outer loop will re-read "iterator" state and
-                                        // attempt to drain from it.
-                                        break;
-                                    }
+                                        iterUpdater.compareAndSet(this, currIter, EmptyIterator.instance()) &&
+                                        sourceSubscription != null) {
+                                    sourceSubscription.request(1);
                                 }
+                                // if the CAS fails or nextIter != currIter, the outer loop will re-read iterator
+                                // and try to emit if items are present and demand allows it.
                             }
                         } finally {
                             // The lock must be released after we interact with the subscription for thread safety

diff --git a/...rrent-api/src/test/java/io/servicetalk/concurrent/api/PublisherConcatMapIterableTest.java b/...rrent-api/src/test/java/io/servicetalk/concurrent/api/PublisherConcatMapIterableTest.java
@@ -428,7 +428,7 @@ void exceptionFromSubscriptionRequestNIsPropagated() {
     }
 
     @Test
-    void testFlatMapConcatIterable() throws Exception {
+    void concurrencyEmitsInOrder() throws Exception {
         try (BlockingIterator<Integer> iterable = fromIterable(() -> range(0, 10000).iterator())
                 .publishOn(Executors.global())
                 .flatMapConcatIterable(Collections::singletonList)