Merge pull request #189 from benjchristensen/polishing

Handful of Improvements

Author: benjchristensen

language-adaptors/rxjava-groovy/src/examples/groovy/rx/lang/groovy/examples/VideoExample.groovy

@@ -108,19 +108,27 @@ Observable getVideoGridForDisplay(userId) {
 Observable<VideoList> getListOfLists(userId) {
     return Observable.create({ observer -> 
         BooleanSubscription subscription = new BooleanSubscription();
-        // this will happen on a separate thread as it requires a network call
-        executor.execute({
-                // simulate network latency
-                Thread.sleep(180);
-                for(i in 0..15) {
-                    if(subscription.isUnsubscribed()) {
-                        break;
+        try {
+            // this will happen on a separate thread as it requires a network call
+            executor.execute({
+                    // simulate network latency
+                    Thread.sleep(180);
+                    for(i in 0..15) {
+                        if(subscription.isUnsubscribed()) {
+                            break;
+                        }
+                        try {
+                            //println("****** emitting list: " + i)
+                            observer.onNext(new VideoList(i))
+                        }catch(Exception e) {
+                            observer.onError(e);
+                        }
                     }
-                    //println("****** emitting list: " + i)
-                    observer.onNext(new VideoList(i))
-                }
-                observer.onCompleted();
-        })
+                    observer.onCompleted();
+            })
+        }catch(Exception e) {
+            observer.onError(e);
+        }
         return subscription;
     })
 }

rxjava-core/src/main/java/rx/Observable.java

@@ -27,7 +27,6 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Before;
@@ -156,11 +155,30 @@ public Subscription subscribe(Observer<T> observer) {
             throw new IllegalStateException("onSubscribe function can not be null.");
             // the subscribe function can also be overridden but generally that's not the appropriate approach so I won't mention that in the exception
         }
-        if (isTrusted) {
-            return onSubscribe.call(observer);
-        } else {
-            AtomicObservableSubscription subscription = new AtomicObservableSubscription();
-            return subscription.wrap(onSubscribe.call(new AtomicObserver<T>(subscription, observer)));
+        try {
+            if (isTrusted) {
+                Subscription s = onSubscribe.call(observer);
+                if (s == null) {
+                    // this generally shouldn't be the case on a 'trusted' onSubscribe but in case it happens
+                    // we want to gracefully handle it the same as AtomicObservableSubscription does
+                    return Subscriptions.empty();
+                } else {
+                    return s;
+                }
+            } else {
+                AtomicObservableSubscription subscription = new AtomicObservableSubscription();
+                return subscription.wrap(onSubscribe.call(new AtomicObserver<T>(subscription, observer)));
+            }
+        } catch (Exception e) {
+            // if an unhandled error occurs executing the onSubscribe we will propagate it
+            try {
+                observer.onError(e);
+            } catch (Exception e2) {
+                // if this happens it means the onError itself failed (perhaps an invalid function implementation)
+                // so we are unable to propagate the error correctly and will just throw
+                throw new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
+            }
+            return Subscriptions.empty();
         }
     }
 
@@ -415,6 +433,18 @@ public void onNext(T args) {
         }
     }
 
+    /**
+     * Invokes an action for each element in the observable sequence, and blocks until the sequence is terminated.
+     * <p>
+     * NOTE: This will block even if the Observable is asynchronous.
+     * <p>
+     * This is similar to {@link #subscribe(Observer)} but blocks. Because it blocks it does not need the {@link Observer#onCompleted()} or {@link Observer#onError(Exception)} methods.
+     * 
+     * @param onNext
+     *            {@link Action1}
+     * @throws RuntimeException
+     *             if error occurs
+     */
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public void forEach(final Object o) {
         if (o instanceof Action1) {
@@ -3405,6 +3435,25 @@ public Boolean call(String s) {
             }));
         }
 
+        @Test
+        public void testOnSubscribeFails() {
+            @SuppressWarnings("unchecked")
+            Observer<String> observer = mock(Observer.class);
+            final RuntimeException re = new RuntimeException("bad impl");
+            Observable<String> o = Observable.create(new Func1<Observer<String>, Subscription>() {
+
+                @Override
+                public Subscription call(Observer<String> t1) {
+                    throw re;
+                }
+
+            });
+            o.subscribe(observer);
+            verify(observer, times(0)).onNext(anyString());
+            verify(observer, times(0)).onCompleted();
+            verify(observer, times(1)).onError(re);
+        }
+
         @Test
         public void testLastEmptyObservable() {
             Observable<String> obs = Observable.toObservable();