OnErrorFlatMap & FlatMap

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

@@ -72,7 +72,15 @@ public void onNext(Observable<? extends T> t) {
         @Override
         public void onError(Throwable e) {
             actual.onError(e);
-            unsubscribe();
+        }
+
+        void errorInner(InnerSubscriber<T> s, Throwable e) {
+            try {
+                WIP_UPDATER.decrementAndGet(this);
+                actual.onError(e);
+            } finally {
+                childrenSubscriptions.remove(s);
+            }
         }
 
         @Override
@@ -114,7 +122,7 @@ public void onNext(T t) {
         public void onError(Throwable e) {
             Exceptions.throwIfFatal(e);
             if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
-                parent.onError(e);
+                parent.errorInner(this, e);
             }
         }
 

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

@@ -107,7 +107,6 @@ public void onError(final Throwable e) {
             if (terminated) {
                 return;
             }
-            terminated = true;
             if (emitting) {
                 if (queue == null) {
                     queue = new FastList();
@@ -121,6 +120,9 @@ public void onError(final Throwable e) {
         }
         drainQueue(list);
         actual.onError(e);
+        synchronized(this) {
+            emitting = false;
+        }
     }
 
     @Override

rxjava-core/src/test/java/rx/internal/operators/OperatorOnErrorFlatMapTest.java

@@ -86,4 +86,33 @@ public Observable<String> call(OnErrorThrowable t) {
         ts.assertReceivedOnNext(Arrays.asList("Value=1", "Error=2", "Error=3", "Error=4", "Error=5", "Value=6"));
     }
 
+    @Test
+    public void testOnErrorFlatMapAfterFlatMap() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.from(1, 2, 3).flatMap(new Func1<Integer, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Integer i) {
+                System.out.println("i: " + i);
+                if (i == 1) {
+                    return Observable.error(new RuntimeException("error"));
+                } else {
+                    return Observable.just(i);
+                }
+            }
+
+        }).onErrorFlatMap(new Func1<OnErrorThrowable, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(OnErrorThrowable t) {
+                System.err.println(t);
+                return Observable.just(-1);
+            }
+
+        }).subscribe(ts);
+        // we won't receive a terminal event so don't wait for one
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Arrays.asList(-1, 2, 3));
+    }
+
 }