Merge pull request #965 from zsxwing/fix-timeout-unsubscribe

BugFix: Chain Subscription in TimeoutSubscriber and SerializedSubscriber

Author: benjchristensen

rxjava-core/src/main/java/rx/observers/SerializedSubscriber.java

@@ -20,6 +20,7 @@ public class SerializedSubscriber<T> extends Subscriber<T> {
     private final Observer<T> s;
 
     public SerializedSubscriber(Subscriber<? super T> s) {
+        super(s);
         this.s = new SerializedObserver<T>(s);
     }
 

rxjava-core/src/main/java/rx/operators/OperatorTimeoutBase.java

@@ -93,6 +93,7 @@ private TimeoutSubscriber(
                 SerializedSubscriber<T> serializedSubscriber,
                 TimeoutStub<T> timeoutStub, SerialSubscription serial,
                 Observable<? extends T> other) {
+            super(serializedSubscriber);
             this.serializedSubscriber = serializedSubscriber;
             this.timeoutStub = timeoutStub;
             this.serial = serial;

rxjava-core/src/test/java/rx/operators/OperatorTimeoutTests.java

@@ -18,6 +18,7 @@
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
+import java.io.IOException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -266,4 +267,89 @@ public void call(Subscriber<? super String> subscriber) {
 
         exit.countDown(); // exit the thread
     }
+
+    @Test
+    public void shouldUnsubscribeFromUnderlyingSubscriptionOnTimeout() throws InterruptedException {
+        // From https://github.com/Netflix/RxJava/pull/951
+        final Subscription s = mock(Subscription.class);
+
+        Observable<String> never = Observable.create(new OnSubscribe<String>() {
+            public void call(Subscriber<? super String> subscriber) {
+                subscriber.add(s);
+            }
+        });
+
+        TestScheduler testScheduler = new TestScheduler();
+        Observable<String> observableWithTimeout = never.timeout(1000, TimeUnit.MILLISECONDS, testScheduler);
+
+        @SuppressWarnings("unchecked")
+        Observer<String> observer = mock(Observer.class);
+        observableWithTimeout.subscribe(observer);
+
+        testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer).onError(isA(TimeoutException.class));
+        inOrder.verifyNoMoreInteractions();
+
+        verify(s, times(1)).unsubscribe();
+    }
+
+    @Test
+    public void shouldUnsubscribeFromUnderlyingSubscriptionOnImmediatelyComplete() {
+        // From https://github.com/Netflix/RxJava/pull/951
+        final Subscription s = mock(Subscription.class);
+
+        Observable<String> immediatelyComplete = Observable.create(new OnSubscribe<String>() {
+            public void call(Subscriber<? super String> subscriber) {
+                subscriber.add(s);
+                subscriber.onCompleted();
+            }
+        });
+
+        TestScheduler testScheduler = new TestScheduler();
+        Observable<String> observableWithTimeout = immediatelyComplete.timeout(1000, TimeUnit.MILLISECONDS,
+                testScheduler);
+
+        @SuppressWarnings("unchecked")
+        Observer<String> observer = mock(Observer.class);
+        observableWithTimeout.subscribe(observer);
+
+        testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+
+        verify(s, times(1)).unsubscribe();
+    }
+
+    @Test
+    public void shouldUnsubscribeFromUnderlyingSubscriptionOnImmediatelyErrored() throws InterruptedException {
+        // From https://github.com/Netflix/RxJava/pull/951
+        final Subscription s = mock(Subscription.class);
+
+        Observable<String> immediatelyError = Observable.create(new OnSubscribe<String>() {
+            public void call(Subscriber<? super String> subscriber) {
+                subscriber.add(s);
+                subscriber.onError(new IOException("Error"));
+            }
+        });
+
+        TestScheduler testScheduler = new TestScheduler();
+        Observable<String> observableWithTimeout = immediatelyError.timeout(1000, TimeUnit.MILLISECONDS,
+                testScheduler);
+
+        @SuppressWarnings("unchecked")
+        Observer<String> observer = mock(Observer.class);
+        observableWithTimeout.subscribe(observer);
+
+        testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer).onError(isA(IOException.class));
+        inOrder.verifyNoMoreInteractions();
+
+        verify(s, times(1)).unsubscribe();
+    }
 }