2.x: make withLatestFrom conditional subscriber, test cold consumption

src/main/java/io/reactivex/internal/operators/flowable/FlowableWithLatestFrom.java

@@ -21,6 +21,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.BiFunction;
 import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.internal.fuseable.ConditionalSubscriber;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.subscribers.SerializedSubscriber;
 
@@ -45,7 +46,8 @@ protected void subscribeActual(Subscriber<? super R> s) {
         source.subscribe(wlf);
     }
 
-    static final class WithLatestFromSubscriber<T, U, R> extends AtomicReference<U> implements FlowableSubscriber<T>, Subscription {
+    static final class WithLatestFromSubscriber<T, U, R> extends AtomicReference<U>
+    implements ConditionalSubscriber<T>, Subscription {
 
         private static final long serialVersionUID = -312246233408980075L;
 
@@ -69,6 +71,13 @@ public void onSubscribe(Subscription s) {
 
         @Override
         public void onNext(T t) {
+            if (!tryOnNext(t)) {
+                s.get().request(1);
+            }
+        }
+
+        @Override
+        public boolean tryOnNext(T t) {
             U u = get();
             if (u != null) {
                 R r;
@@ -78,11 +87,12 @@ public void onNext(T t) {
                     Exceptions.throwIfFatal(e);
                     cancel();
                     actual.onError(e);
-                    return;
+                    return false;
                 }
                 actual.onNext(r);
-            } else{
-                request(1);
+                return true;
+            } else {
+                return false;
             }
         }
 

src/main/java/io/reactivex/internal/operators/flowable/FlowableWithLatestFromMany.java

@@ -23,6 +23,7 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.internal.fuseable.ConditionalSubscriber;
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -95,7 +96,7 @@ protected void subscribeActual(Subscriber<? super R> s) {
 
     static final class WithLatestFromSubscriber<T, R>
     extends AtomicInteger
-    implements FlowableSubscriber<T>, Subscription {
+    implements ConditionalSubscriber<T>, Subscription {
 
         private static final long serialVersionUID = 1577321883966341961L;
 
@@ -147,8 +148,15 @@ public void onSubscribe(Subscription s) {
 
         @Override
         public void onNext(T t) {
+            if (!tryOnNext(t) && !done) {
+                s.get().request(1);
+            }
+        }
+
+        @Override
+        public boolean tryOnNext(T t) {
             if (done) {
-                return;
+                return false;
             }
             AtomicReferenceArray<Object> ara = values;
             int n = ara.length();
@@ -159,8 +167,7 @@ public void onNext(T t) {
                 Object o = ara.get(i);
                 if (o == null) {
                     // somebody hasn't signalled yet, skip this T
-                    s.get().request(1);
-                    return;
+                    return false;
                 }
                 objects[i + 1] = o;
             }
@@ -173,10 +180,11 @@ public void onNext(T t) {
                 Exceptions.throwIfFatal(ex);
                 cancel();
                 onError(ex);
-                return;
+                return false;
             }
 
             HalfSerializer.onNext(actual, v, this, error);
+            return true;
         }
 
         @Override

src/test/java/io/reactivex/internal/operators/flowable/FlowableWithLatestFromTest.java

@@ -726,7 +726,7 @@ public void zeroOtherCombinerReturnsNull() {
     }
 
     @Test
-    public void testSingleRequestNotForgottenWhenNoData() {
+    public void singleRequestNotForgottenWhenNoData() {
         PublishProcessor<Integer> source = PublishProcessor.create();
         PublishProcessor<Integer> other = PublishProcessor.create();
 
@@ -750,4 +750,30 @@ public void testSingleRequestNotForgottenWhenNoData() {
 
         ts.assertValue((2 << 8) + 1);
     }
+
+    @Test
+    public void coldSourceConsumedWithoutOther() {
+        Flowable.range(1, 10).withLatestFrom(Flowable.never(),
+        new BiFunction<Integer, Object, Object>() {
+            @Override
+            public Object apply(Integer a, Object b) throws Exception {
+                return a;
+            }
+        })
+        .test(1)
+        .assertResult();
+    }
+
+    @Test
+    public void coldSourceConsumedWithoutManyOthers() {
+        Flowable.range(1, 10).withLatestFrom(Flowable.never(), Flowable.never(), Flowable.never(),
+        new Function4<Integer, Object, Object, Object, Object>() {
+            @Override
+            public Object apply(Integer a, Object b, Object c, Object d) throws Exception {
+                return a;
+            }
+        })
+        .test(1)
+        .assertResult();
+    }
 }