Merge pull request ReactiveX#2156 from zsxwing/map-swallow-fatal-exce…

…ptions

Fix the issue that map may swallow fatal exceptions

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

@@ -17,6 +17,7 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.exceptions.OnErrorThrowable;
 import rx.functions.Func1;
 
@@ -53,6 +54,7 @@ public void onNext(T t) {
                 try {
                     o.onNext(transformer.call(t));
                 } catch (Throwable e) {
+                    Exceptions.throwIfFatal(e);
                     onError(OnErrorThrowable.addValueAsLastCause(e, t));
                 }
             }

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

@@ -319,4 +319,24 @@ private static Map<String, String> getMap(String prefix) {
         m.put("lastName", prefix + "Last");
         return m;
     }
+
+    @Test(expected = OnErrorNotImplementedException.class)
+    public void testShouldNotSwallowOnErrorNotImplementedException() {
+        Observable.just("a", "b").flatMap(new Func1<String, Observable<String>>() {
+            @Override
+            public Observable<String> call(String s) {
+                return Observable.just(s + "1", s + "2");
+            }
+        }).flatMap(new Func1<String, Observable<String>>() {
+            @Override
+            public Observable<String> call(String s) {
+                return Observable.error(new Exception("test"));
+            }
+        }).forEach(new Action1<String>() {
+            @Override
+            public void call(String s) {
+                System.out.println(s);
+            }
+        });
+    }
 }