Merge pull request #1367 from zsxwing/swallow-error

Fix the bug that 'flatMap' swallows OnErrorNotImplementedException

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/operators/OperatorMerge.java

@@ -20,6 +20,7 @@
 import rx.Observable;
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.observers.SerializedSubscriber;
 import rx.subscriptions.CompositeSubscription;
 
@@ -111,6 +112,7 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable e) {
+            Exceptions.throwIfFatal(e);
             if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
                 parent.onError(e);
             }

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

@@ -16,6 +16,7 @@
 package rx.observers;
 
 import rx.Observer;
+import rx.exceptions.Exceptions;
 
 /**
  * Enforces single-threaded, serialized, ordered execution of {@link #onNext}, {@link #onCompleted}, and
@@ -100,6 +101,7 @@ public void onCompleted() {
 
     @Override
     public void onError(final Throwable e) {
+        Exceptions.throwIfFatal(e);
         FastList list;
         synchronized (this) {
             if (terminated) {

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -46,6 +46,7 @@
 import org.mockito.MockitoAnnotations;
 
 import rx.Observable.OnSubscribe;
+import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action1;
 import rx.functions.Action2;
 import rx.functions.Func1;
@@ -1016,4 +1017,14 @@ public void testAmbWith() {
         ts.assertReceivedOnNext(Arrays.asList(1));
     }
 
+    @Test(expected = OnErrorNotImplementedException.class)
+    public void testSubscribeWithoutOnError() {
+        Observable<String> o = Observable.from("a", "b").flatMap(new Func1<String, Observable<String>>() {
+            @Override
+            public Observable<String> call(String s) {
+                return Observable.error(new Exception("test"));
+            }
+        });
+        o.subscribe();
+    }
 }