2.x: fix timeout with fallback not cancelling the main source (#4945)

Author: akarnokd

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

@@ -137,7 +137,7 @@ static final class MulticastProcessor<T> extends Flowable<T> implements Subscrib
         final AtomicReference<MulticastSubscription<T>[]> subscribers;
 
         final int prefetch;
-        
+
         final int limit;
 
         final boolean delayError;
@@ -150,7 +150,7 @@ static final class MulticastProcessor<T> extends Flowable<T> implements Subscrib
 
         volatile boolean done;
         Throwable error;
-        
+
         int consumed;
 
         @SuppressWarnings("unchecked")
@@ -319,11 +319,11 @@ void drain() {
             int missed = 1;
 
             SimpleQueue<T> q = queue;
-            
+
             int upstreamConsumed = consumed;
             int localLimit = limit;
             boolean canRequest = sourceMode != QueueSubscription.SYNC;
-            
+
             for (;;) {
                 MulticastSubscription<T>[] array = subscribers.get();
 

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

@@ -174,6 +174,7 @@ public void onComplete() {
         public void dispose() {
             worker.dispose();
             DisposableHelper.dispose(timer);
+            s.cancel();
         }
 
         @Override

src/main/java/io/reactivex/internal/operators/observable/ObservableTimeoutTimed.java

@@ -174,6 +174,7 @@ public void onComplete() {
         public void dispose() {
             worker.dispose();
             DisposableHelper.dispose(this);
+            s.dispose();
         }
 
         @Override

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

@@ -473,7 +473,7 @@ public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
                             public boolean test(Integer w) throws Exception {
                                 return w % 2 == 0;
                             }
-                        }), 
+                        }),
                         v.filter(new Predicate<Integer>() {
                             @Override
                             public boolean test(Integer w) throws Exception {
@@ -500,7 +500,7 @@ public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
                             public boolean test(Integer w) throws Exception {
                                 return w % 2 == 0;
                             }
-                        }), 
+                        }),
                         v.filter(new Predicate<Integer>() {
                             @Override
                             public boolean test(Integer w) throws Exception {
@@ -528,7 +528,7 @@ public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
                             public boolean test(Integer w) throws Exception {
                                 return w % 2 == 0;
                             }
-                        }), 
+                        }),
                         v.filter(new Predicate<Integer>() {
                             @Override
                             public boolean test(Integer w) throws Exception {

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

@@ -483,4 +483,39 @@ protected void subscribeActual(Subscriber<? super Integer> observer) {
         }
     }
 
+
+    @Test
+    public void timedTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps.timeout(1, TimeUnit.DAYS)
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void timedFallbackTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps.timeout(1, TimeUnit.DAYS, Flowable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
 }

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

@@ -13,7 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
@@ -509,4 +509,40 @@ protected void subscribeActual(Subscriber<? super Integer> observer) {
         .test()
         .assertResult(1);
     }
+
+    @Test
+    public void selectorTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps
+        .timeout(Functions.justFunction(Flowable.never()))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void selectorFallbackTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps
+        .timeout(Functions.justFunction(Flowable.never()), Flowable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutTests.java

@@ -481,4 +481,38 @@ protected void subscribeActual(Observer<? super Integer> observer) {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void timedTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps.timeout(1, TimeUnit.DAYS)
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void timedFallbackTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps.timeout(1, TimeUnit.DAYS, Observable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutWithSelectorTest.java

@@ -13,7 +13,7 @@
 
 package io.reactivex.internal.operators.observable;
 
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
@@ -510,4 +510,40 @@ protected void subscribeActual(Observer<? super Integer> observer) {
         .test()
         .assertResult(1);
     }
+
+    @Test
+    public void selectorTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps
+        .timeout(Functions.justFunction(Observable.never()))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void selectorFallbackTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps
+        .timeout(Functions.justFunction(Observable.never()), Observable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
 }