Merge pull request #1813 from zsxwing/issue1812

Fix issue #1812 that zip may swallow requests

Author: benjchristensen

src/main/java/rx/internal/operators/OperatorZip.java

@@ -227,7 +227,7 @@ void tick() {
             if (COUNTER_UPDATER.getAndIncrement(this) == 0) {
                 do {
                     // we only emit if requested > 0
-                    if (requested.get() > 0) {
+                    while (requested.get() > 0) {
                         final Object[] vs = new Object[observers.length];
                         boolean allHaveValues = true;
                         for (int i = 0; i < observers.length; i++) {
@@ -279,6 +279,8 @@ void tick() {
                                 }
                                 emitted = 0;
                             }
+                        } else {
+                            break;
                         }
                     }
                 } while (COUNTER_UPDATER.decrementAndGet(this) > 0);

src/test/java/rx/internal/operators/OperatorZipTest.java

@@ -1217,4 +1217,30 @@ public void run() {
 
         });
     }
+
+    @Test(timeout = 30000)
+    public void testIssue1812() {
+        // https://github.com/ReactiveX/RxJava/issues/1812
+        Observable<Integer> zip1 = Observable.zip(Observable.range(0, 1026), Observable.range(0, 1026),
+                new Func2<Integer, Integer, Integer>() {
+
+                    @Override
+                    public Integer call(Integer i1, Integer i2) {
+                        return i1 + i2;
+                    }
+                });
+        Observable<Integer> zip2 = Observable.zip(zip1, Observable.range(0, 1026),
+                new Func2<Integer, Integer, Integer>() {
+
+                    @Override
+                    public Integer call(Integer i1, Integer i2) {
+                        return i1 + i2;
+                    }
+                });
+        List<Integer> expected = new ArrayList<Integer>();
+        for (int i = 0; i < 1026; i++) {
+            expected.add(i * 3);
+        }
+        assertEquals(expected, zip2.toList().toBlocking().single());
+    }
 }