Zip: emit onCompleted without waiting for request + avoid re-reading

fields

Author: akarnokd

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

@@ -223,17 +223,21 @@ public void start(@SuppressWarnings("rawtypes") Observable[] os, AtomicLong requ
          */
         @SuppressWarnings("unchecked")
         void tick() {
+            final Object[] observers = this.observers;
             if (observers == null) {
                 // nothing yet to do (initial request from Producer)
                 return;
             }
             if (COUNTER_UPDATER.getAndIncrement(this) == 0) {
+                final int length = observers.length;
+                final Observer<? super R> child = this.child;
+                final AtomicLong requested = this.requested;
                 do {
-                    // we only emit if requested > 0
-                    while (requested.get() > 0) {
-                        final Object[] vs = new Object[observers.length];
+                    while (true) {
+                        // peek for a potential onCompleted event
+                        final Object[] vs = new Object[length];
                         boolean allHaveValues = true;
-                        for (int i = 0; i < observers.length; i++) {
+                        for (int i = 0; i < length; i++) {
                             RxRingBuffer buffer = ((InnerSubscriber) observers[i]).items;
                             Object n = buffer.peek();
 
@@ -252,7 +256,8 @@ void tick() {
                                 vs[i] = buffer.getValue(n);
                             }
                         }
-                        if (allHaveValues) {
+                        // we only emit if requested > 0 and have all values available
+                        if (requested.get() > 0 && allHaveValues) {
                             try {
                                 // all have something so emit
                                 child.onNext(zipFunction.call(vs));

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

@@ -36,6 +36,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import junit.framework.Assert;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -1266,4 +1268,43 @@ public void onNext(Integer t) {
         ts.assertTerminalEvent();
         ts.assertReceivedOnNext(Arrays.asList(11, 22));
     }
+    @Test(timeout = 10000)
+    public void testZipRace() {
+        Observable<Integer> src = Observable.just(1).subscribeOn(Schedulers.computation());
+        for (int i = 0; i < 500000; i++) {
+            int value = Observable.zip(src, src, new Func2<Integer, Integer, Integer>() {
+                @Override
+                public Integer call(Integer t1, Integer t2) {
+                    return t1 + t2 * 10;
+                }
+            }).toBlocking().singleOrDefault(0);
+            
+            Assert.assertEquals(11, value);
+        }
+    }
+    /** 
+     * Request only a single value and don't wait for another request just
+     * to emit an onCompleted.
+     */
+    @Test
+    public void testZipRequest1() {
+        Observable<Integer> src = Observable.just(1).subscribeOn(Schedulers.computation());
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onStart() {
+                requestMore(1);
+            }
+        };
+        
+        Observable.zip(src, src, new Func2<Integer, Integer, Integer>() {
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return t1 + t2 * 10;
+            }
+        }).subscribe(ts);
+        
+        ts.awaitTerminalEvent(1, TimeUnit.SECONDS);
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Arrays.asList(11));
+    }
 }