Unsubscribe when thread is interrupted

Author: Roman Mazur

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

@@ -23,6 +23,7 @@
 import rx.Notification;
 import rx.Observable;
 import rx.Subscriber;
+import rx.Subscription;
 import rx.exceptions.Exceptions;
 
 /**
@@ -49,7 +50,7 @@ public static <T> Iterable<T> latest(final Observable<? extends T> source) {
             @Override
             public Iterator<T> iterator() {
                 LatestObserverIterator<T> lio = new LatestObserverIterator<T>();
-                source.materialize().subscribe(lio);
+                lio.subscription = source.materialize().subscribe(lio);
                 return lio;
             }
         };
@@ -60,6 +61,7 @@ static final class LatestObserverIterator<T> extends Subscriber<Notification<? e
         final Semaphore notify = new Semaphore(0);
         // observer's notification
         volatile Notification<? extends T> value;
+        volatile Subscription subscription;
         /** Updater for the value field. */
         @SuppressWarnings("rawtypes")
         static final AtomicReferenceFieldUpdater<LatestObserverIterator, Notification> REFERENCE_UPDATER
@@ -96,6 +98,7 @@ public boolean hasNext() {
                     try {
                         notify.acquire();
                     } catch (InterruptedException ex) {
+                        subscription.unsubscribe();
                         Thread.currentThread().interrupt();
                         iNotif = Notification.createOnError(ex);
                         throw Exceptions.propagate(ex);

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

@@ -24,6 +24,7 @@
 import rx.Notification;
 import rx.Observable;
 import rx.Subscriber;
+import rx.Subscription;
 import rx.exceptions.Exceptions;
 
 /**
@@ -65,6 +66,7 @@ public Iterator<T> iterator() {
         private boolean isNextConsumed = true;
         private Throwable error = null;
         private boolean started = false;
+        private Subscription subscription;
 
         private NextIterator(Observable<? extends T> items, NextObserver<T> observer) {
             this.items = items;
@@ -96,7 +98,7 @@ private boolean moveToNext() {
                     started = true;
                     // if not started, start now
                     observer.setWaiting(1);
-                    items.materialize().subscribe(observer);
+                    subscription = items.materialize().subscribe(observer);
                 }
 
                 Notification<? extends T> nextNotification = observer.takeNext();
@@ -117,6 +119,8 @@ private boolean moveToNext() {
                 }
                 throw new IllegalStateException("Should not reach here");
             } catch (InterruptedException e) {
+                // Subscription is always created before the first observer.takeNext().
+                subscription.unsubscribe();
                 Thread.currentThread().interrupt();
                 error = e;
                 throw Exceptions.propagate(error);

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

@@ -23,6 +23,7 @@
 import rx.Notification;
 import rx.Observable;
 import rx.Subscriber;
+import rx.Subscription;
 import rx.exceptions.Exceptions;
 
 /**
@@ -49,7 +50,7 @@ public static <T> Iterator<T> toIterator(Observable<? extends T> source) {
         final BlockingQueue<Notification<? extends T>> notifications = new LinkedBlockingQueue<Notification<? extends T>>();
 
         // using subscribe instead of unsafeSubscribe since this is a BlockingObservable "final subscribe"
-        source.materialize().subscribe(new Subscriber<Notification<? extends T>>() {
+        final Subscription subscription = source.materialize().subscribe(new Subscriber<Notification<? extends T>>() {
             @Override
             public void onCompleted() {
                 // ignore
@@ -94,6 +95,7 @@ private Notification<? extends T> take() {
                 try {
                     return notifications.take();
                 } catch (InterruptedException e) {
+                    subscription.unsubscribe();
                     throw Exceptions.propagate(e);
                 }
             }

src/main/java/rx/observables/BlockingObservable.java

@@ -23,6 +23,7 @@
 
 import rx.Observable;
 import rx.Subscriber;
+import rx.Subscription;
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.functions.Functions;
@@ -95,7 +96,7 @@ public void forEach(final Action1<? super T> onNext) {
          * Use 'subscribe' instead of 'unsafeSubscribe' for Rx contract behavior
          * as this is the final subscribe in the chain.
          */
-        o.subscribe(new Subscriber<T>() {
+        Subscription subscription = o.subscribe(new Subscriber<T>() {
             @Override
             public void onCompleted() {
                 latch.countDown();
@@ -124,6 +125,7 @@ public void onNext(T args) {
         try {
             latch.await();
         } catch (InterruptedException e) {
+            subscription.unsubscribe();
             // set the interrupted flag again so callers can still get it
             // for more information see https://github.com/ReactiveX/RxJava/pull/147#issuecomment-13624780
             Thread.currentThread().interrupt();
@@ -450,7 +452,7 @@ private T blockForSingle(final Observable<? extends T> observable) {
         final AtomicReference<Throwable> returnException = new AtomicReference<Throwable>();
         final CountDownLatch latch = new CountDownLatch(1);
 
-        observable.subscribe(new Subscriber<T>() {
+        Subscription subscription = observable.subscribe(new Subscriber<T>() {
             @Override
             public void onCompleted() {
                 latch.countDown();
@@ -471,6 +473,7 @@ public void onNext(final T item) {
         try {
             latch.await();
         } catch (InterruptedException e) {
+            subscription.unsubscribe();
             Thread.currentThread().interrupt();
             throw new RuntimeException("Interrupted while waiting for subscription to complete.", e);
         }

src/test/java/rx/observables/BlockingObservableTest.java

@@ -357,7 +357,7 @@ public void testFirstOrDefault() {
 
     @Test
     public void testFirstOrDefaultWithEmpty() {
-        BlockingObservable<String> observable = BlockingObservable.from(Observable.<String> empty());
+        BlockingObservable<String> observable = BlockingObservable.from(Observable.<String>empty());
         assertEquals("default", observable.firstOrDefault("default"));
     }
 
@@ -410,4 +410,136 @@ public void call() {
         }
         assertTrue("Timeout means `unsubscribe` is not called", unsubscribe.await(30, TimeUnit.SECONDS));
     }
+
+    @Test
+    public void testUnsubscribeFromSingleWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("single()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.single();
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromForEachWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("forEach()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.forEach(new Action1<Void>() {
+                    @Override
+                    public void call(final Void aVoid) {
+                        // nothing
+                    }
+                });
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromFirstWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("first()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.first();
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromLastWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("last()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.last();
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromLatestWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("latest()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.latest().iterator().next();
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromNextWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("next()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.next().iterator().next();
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromGetIteratorWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("getIterator()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.getIterator().next();
+            }
+        });
+    }
+
+    @Test
+    public void testUnsubscribeFromToIterableWhenInterrupted() throws InterruptedException {
+        new InterruptionTests().assertUnsubscribeIsInvoked("toIterable()", new Action1<BlockingObservable<Void>>() {
+            @Override
+            public void call(final BlockingObservable<Void> o) {
+                o.toIterable().iterator().next();
+            }
+        });
+    }
+
+    /** Utilities set for interruption behaviour tests. */
+    private static class InterruptionTests {
+
+        private boolean isUnSubscribed;
+        private RuntimeException error;
+        private CountDownLatch latch = new CountDownLatch(1);
+
+        private Observable<Void> createObservable() {
+            return Observable.<Void>never().doOnUnsubscribe(new Action0() {
+                @Override
+                public void call() {
+                    isUnSubscribed = true;
+                }
+            });
+        }
+
+        private void startBlockingAndInterrupt(final Action1<BlockingObservable<Void>> blockingAction) {
+            Thread subscriptionThread = new Thread() {
+                @Override
+                public void run() {
+                    try {
+                        blockingAction.call(createObservable().toBlocking());
+                    } catch (RuntimeException e) {
+                        if (!(e.getCause() instanceof InterruptedException)) {
+                            error = e;
+                        }
+                    }
+                    latch.countDown();
+                }
+            };
+            subscriptionThread.start();
+            subscriptionThread.interrupt();
+        }
+
+        void assertUnsubscribeIsInvoked(final String method, final Action1<BlockingObservable<Void>> blockingAction)
+            throws InterruptedException {
+            startBlockingAndInterrupt(blockingAction);
+            assertTrue("Timeout means interruption is not performed", latch.await(30, TimeUnit.SECONDS));
+            if (error != null) {
+                throw error;
+            }
+            assertTrue("'unsubscribe' is not invoked when thread is interrupted for " + method, isUnSubscribed);
+        }
+
+    }
+
 }