No InterruptedException with synchronous BlockingObservable

Author: ypresto

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

@@ -93,6 +93,10 @@ public T next() {
 
             private Notification<? extends T> take() {
                 try {
+                    Notification<? extends T> poll = notifications.poll();
+                    if (poll != null) {
+                        return poll;
+                    }
                     return notifications.take();
                 } catch (InterruptedException e) {
                     subscription.unsubscribe();

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

@@ -123,17 +123,7 @@ public void onNext(T args) {
                 onNext.call(args);
             }
         });
-        // block until the subscription completes and then return
-        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();
-            // using Runtime so it is not checked
-            throw new RuntimeException("Interrupted while waiting for subscription to complete.", e);
-        }
+        awaitForComplete(latch, subscription);
 
         if (exceptionFromOnError.get() != null) {
             if (exceptionFromOnError.get() instanceof RuntimeException) {
@@ -456,14 +446,7 @@ public void onNext(final T item) {
                 returnItem.set(item);
             }
         });
-
-        try {
-            latch.await();
-        } catch (InterruptedException e) {
-            subscription.unsubscribe();
-            Thread.currentThread().interrupt();
-            throw new RuntimeException("Interrupted while waiting for subscription to complete.", e);
-        }
+        awaitForComplete(latch, subscription);
 
         if (returnException.get() != null) {
             if (returnException.get() instanceof RuntimeException) {
@@ -475,4 +458,23 @@ public void onNext(final T item) {
 
         return returnItem.get();
     }
+
+    private void awaitForComplete(CountDownLatch latch, Subscription subscription) {
+        if (latch.getCount() == 0) {
+            // Synchronous observable completes before awaiting for it.
+            // Skip await so InterruptedException will never be thrown.
+            return;
+        }
+        // block until the subscription completes and then return
+        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();
+            // using Runtime so it is not checked
+            throw new RuntimeException("Interrupted while waiting for subscription to complete.", e);
+        }
+    }
 }

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

@@ -15,11 +15,8 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import static rx.internal.operators.BlockingOperatorNext.next;
+import org.junit.Assert;
+import org.junit.Test;
 
 import java.util.Iterator;
 import java.util.NoSuchElementException;
@@ -28,19 +25,21 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.Assert;
-import org.junit.Test;
-
 import rx.Observable;
 import rx.Subscriber;
 import rx.exceptions.TestException;
-import rx.internal.operators.BlockingOperatorNext;
 import rx.observables.BlockingObservable;
 import rx.schedulers.Schedulers;
 import rx.subjects.BehaviorSubject;
 import rx.subjects.PublishSubject;
 import rx.subjects.Subject;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static rx.internal.operators.BlockingOperatorNext.next;
+
 public class BlockingOperatorNextTest {
 
     private void fireOnNextInNewThread(final Subject<String, String> o, final String value) {
@@ -83,6 +82,13 @@ public void testNext() {
         assertTrue(it.hasNext());
         assertEquals("two", it.next());
 
+        fireOnNextInNewThread(obs, "three");
+        try {
+            assertEquals("three", it.next());
+        } catch (NoSuchElementException e) {
+            fail("Calling next() without hasNext() should wait for next fire");
+        }
+
         obs.onCompleted();
         assertFalse(it.hasNext());
         try {