Merge pull request #2590 from akarnokd/ZipHangFix

Zip: fixed unbounded downstream requesting above Long.MAX_VALUE
ReactiveX · Feb 3, 2015 · 43a912f · 43a912f
2 parents 0cecfc5 + 01e97fc
commit 43a912f
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 2 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorZip.java b/src/main/java/rx/internal/operators/OperatorZip.java
@@ -117,6 +117,7 @@ public Subscriber<? super Observable[]> call(final Subscriber<? super R> child)
         return subscriber;
     }
 
+    @SuppressWarnings("rawtypes")
     private final class ZipSubscriber extends Subscriber<Observable[]> {
 
         final Subscriber<? super R> child;
@@ -158,7 +159,8 @@ public void onNext(Observable[] observables) {
     }
 
     private static final class ZipProducer<R> extends AtomicLong implements Producer {
-
+        /** */
+        private static final long serialVersionUID = -1216676403723546796L;
         private Zip<R> zipper;
 
         public ZipProducer(Zip<R> zipper) {
@@ -167,7 +169,7 @@ public ZipProducer(Zip<R> zipper) {
 
         @Override
         public void request(long n) {
-            addAndGet(n);
+            BackpressureUtils.getAndAddRequest(this, n);
             // try and claim emission if no other threads are doing so
             zipper.tick();
         }
@@ -179,6 +181,7 @@ private static final class Zip<R> {
         private final FuncN<? extends R> zipFunction;
         private final CompositeSubscription childSubscription = new CompositeSubscription();
 
+        @SuppressWarnings("unused")
         volatile long counter;
         @SuppressWarnings("rawtypes")
         static final AtomicLongFieldUpdater<Zip> COUNTER_UPDATER = AtomicLongFieldUpdater.newUpdater(Zip.class, "counter");

diff --git a/src/test/java/rx/internal/operators/OperatorZipTest.java b/src/test/java/rx/internal/operators/OperatorZipTest.java
@@ -1243,4 +1243,27 @@ public Integer call(Integer i1, Integer i2) {
         }
         assertEquals(expected, zip2.toList().toBlocking().single());
     }
+    @Test
+    public void testUnboundedDownstreamOverrequesting() {
+        Observable<Integer> source = Observable.range(1, 2).zipWith(Observable.range(1, 2), new Func2<Integer, Integer, Integer>() {
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return t1 + 10 * t2;
+            }
+        });
+
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onNext(Integer t) {
+                super.onNext(t);
+                requestMore(5);
+            }
+        };
+
+        source.subscribe(ts);
+
+        ts.assertNoErrors();
+        ts.assertTerminalEvent();
+        ts.assertReceivedOnNext(Arrays.asList(11, 22));
+    }
 }