Merge pull request #2951 from davidmoten/concat-request-overflow

OperatorConcat - prevent request overflow and fix race condition
ReactiveX · May 15, 2015 · 425a6f4 · 425a6f4
2 parents 1b25f07 + bad4d40
commit 425a6f4
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 1 deletion.
diff --git a/src/main/java/rx/internal/operators/OperatorConcat.java b/src/main/java/rx/internal/operators/OperatorConcat.java
@@ -115,7 +115,7 @@ public void onStart() {
         private void requestFromChild(long n) {
             // we track 'requested' so we know whether we should subscribe the next or not
             ConcatInnerSubscriber<T> actualSubscriber = currentSubscriber;
-            if (REQUESTED_UPDATER.getAndAdd(this, n) == 0) {
+            if (n > 0 && BackpressureUtils.getAndAddRequest(REQUESTED_UPDATER, this, n) == 0) {
                 if (actualSubscriber == null && wip > 0) {
                     // this means we may be moving from one subscriber to another after having stopped processing
                     // so need to kick off the subscribe via this request notification

diff --git a/src/test/java/rx/internal/operators/OperatorConcatTest.java b/src/test/java/rx/internal/operators/OperatorConcatTest.java
@@ -16,6 +16,7 @@
 package rx.internal.operators;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
@@ -35,7 +36,9 @@
 import org.mockito.InOrder;
 
 import rx.Observable.OnSubscribe;
+import rx.Scheduler.Worker;
 import rx.*;
+import rx.functions.Action0;
 import rx.functions.Func1;
 import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
@@ -766,4 +769,30 @@ public void onError(Throwable e) {
 
         assertEquals(n, counter.get());
     }
+
+    @Test
+    public void testRequestOverflowDoesNotStallStream() {
+        Observable<Integer> o1 = Observable.just(1,2,3);
+        Observable<Integer> o2 = Observable.just(4,5,6);
+        final AtomicBoolean completed = new AtomicBoolean(false);
+        o1.concatWith(o2).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onCompleted() {
+                completed.set(true);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                request(2);
+            }});
+
+        assertTrue(completed.get());
+    }
+
 }