Merge pull request #3493 from akarnokd/ZipBackpressureFix1x

1.x: fix for zip(Obs<Obs<T>>) backpressure problem

Author: akarnokd

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

@@ -111,8 +111,11 @@ public OperatorZip(Func9 f) {
     public Subscriber<? super Observable[]> call(final Subscriber<? super R> child) {
         final Zip<R> zipper = new Zip<R>(child, zipFunction);
         final ZipProducer<R> producer = new ZipProducer<R>(zipper);
-        child.setProducer(producer);
         final ZipSubscriber subscriber = new ZipSubscriber(child, zipper, producer);
+
+        child.add(subscriber);
+        child.setProducer(producer);
+        
         return subscriber;
     }
 
@@ -124,7 +127,6 @@ private final class ZipSubscriber extends Subscriber<Observable[]> {
         final ZipProducer<R> producer;
 
         public ZipSubscriber(Subscriber<? super R> child, Zip<R> zipper, ZipProducer<R> producer) {
-            super(child);
             this.child = child;
             this.zipper = zipper;
             this.producer = producer;

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

@@ -1308,4 +1308,32 @@ public Integer call(Integer t1, Integer t2) {
         ts.assertNoErrors();
         ts.assertReceivedOnNext(Arrays.asList(11));
     }
+    
+    @SuppressWarnings("cast")
+    @Test
+    public void testZipObservableObservableBackpressure() {
+        @SuppressWarnings("unchecked")
+        Observable<Integer>[] osArray = new Observable[] { 
+                Observable.range(0, 10), 
+                Observable.range(0, 10) 
+        };
+        
+        Observable<Observable<Integer>> os = (Observable<Observable<Integer>>) Observable.from(osArray);
+        Observable<Integer> o1 = Observable.zip(os, new FuncN<Integer>() {
+            @Override
+            public Integer call(Object... a) {
+                return 0;
+            }
+        });
+        
+        TestSubscriber<Integer> sub1 = TestSubscriber.create(5);
+        
+        o1.subscribe(sub1);
+
+        sub1.requestMore(5);
+
+        sub1.assertValueCount(10);
+        sub1.assertNoErrors();
+        sub1.assertCompleted();
+    }
 }