Merge pull request #2543 from davidmoten/merge-request-overflow

OperatorMerge handle request overflow
ReactiveX · Jan 28, 2015 · 19a66b7 · 19a66b7
2 parents 18def88 + 5348ebf
commit 19a66b7
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 1 deletion.
diff --git a/src/main/java/rx/internal/operators/OperatorMerge.java b/src/main/java/rx/internal/operators/OperatorMerge.java
@@ -545,7 +545,16 @@ public void request(long n) {
             if (n == Long.MAX_VALUE) {
                 requested = Long.MAX_VALUE;
             } else {
-                REQUESTED.getAndAdd(this, n);
+                // add n to requested but check for overflow
+                while (true) {
+                    long current = REQUESTED.get(this);
+                    long next = current + n;
+                    //check for overflow
+                    if (next < 0)
+                        next = Long.MAX_VALUE;
+                    if (REQUESTED.compareAndSet(this, current, next))
+                        break;
+                }
                 if (ms.drainQueuesIfNeeded()) {
                     boolean sendComplete = false;
                     synchronized (ms) {

diff --git a/src/test/java/rx/internal/operators/OperatorMergeTest.java b/src/test/java/rx/internal/operators/OperatorMergeTest.java
@@ -1183,6 +1183,38 @@ public void call() {
             assertTrue(a);
         //}
     }
+
+    @Test
+    public void testMergeRequestOverflow() throws InterruptedException {
+        //do a non-trivial merge so that future optimisations with EMPTY don't invalidate this test
+        Observable<Integer> o = Observable.from(Arrays.asList(1,2)).mergeWith(Observable.from(Arrays.asList(3,4)));
+        final int expectedCount = 4;
+        final CountDownLatch latch = new CountDownLatch(expectedCount);
+        o.subscribeOn(Schedulers.computation()).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(1);
+            }
+
+            @Override
+            public void onCompleted() {
+                //ignore
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                throw new RuntimeException(e);
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                latch.countDown();
+                request(2);
+                request(Long.MAX_VALUE-1);
+            }});
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+    }
 
     private static Action1<Integer> printCount() {
         return new Action1<Integer>() {