Merge pull request #3528 from zsxwing/zipIter-delay

1.x: Avoid to call next when Iterator is drained

Author: akarnokd

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

@@ -46,23 +46,30 @@ public Subscriber<? super T1> call(final Subscriber<? super R> subscriber) {
             return Subscribers.empty();
         }
         return new Subscriber<T1>(subscriber) {
-            boolean once;
+            boolean done;
             @Override
             public void onCompleted() {
-                if (once) {
+                if (done) {
                     return;
                 }
-                once = true;
+                done = true;
                 subscriber.onCompleted();
             }
 
             @Override
             public void onError(Throwable e) {
+                if (done) {
+                    return;
+                }
+                done = true;
                 subscriber.onError(e);
             }
 
             @Override
             public void onNext(T1 t) {
+                if (done) {
+                    return;
+                }
                 try {
                     subscriber.onNext(zipFunction.call(t, iterator.next()));
                     if (!iterator.hasNext()) {

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

@@ -23,6 +23,7 @@
 
 import java.util.Arrays;
 import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import static org.junit.Assert.assertEquals;
 
@@ -37,6 +38,8 @@
 import rx.functions.Func1;
 import rx.functions.Func2;
 import rx.functions.Func3;
+import rx.observers.TestSubscriber;
+import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 
 public class OperatorZipIterableTest {
@@ -378,4 +381,22 @@ public String call(Integer t1) {
 
         assertEquals(2, squareStr.counter.get());
     }
+
+    @Test
+    public void testZipIterableWithDelay() {
+        TestScheduler scheduler = new TestScheduler();
+        Observable<Integer> o = Observable.just(1, 2).zipWith(Arrays.asList(1), new Func2<Integer, Integer, Integer>() {
+            @Override
+            public Integer call(Integer v1, Integer v2) {
+                return v1;
+            }
+        }).delay(500, TimeUnit.MILLISECONDS, scheduler);
+
+        TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>();
+        o.subscribe(subscriber);
+        scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS);
+        subscriber.assertValue(1);
+        subscriber.assertNoErrors();
+        subscriber.assertCompleted();
+    }
 }