1.x: DoAfterTerminate handle if action throws

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

@@ -17,7 +17,9 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.functions.Action0;
+import rx.plugins.RxJavaPlugins;
 
 /**
  * Registers an action to be called after an Observable invokes {@code onComplete} or {@code onError}.
@@ -53,7 +55,7 @@ public void onError(Throwable e) {
                 try {
                     child.onError(e);
                 } finally {
-                    action.call();
+                    callAction();
                 }
             }
 
@@ -62,7 +64,16 @@ public void onCompleted() {
                 try {
                     child.onCompleted();
                 } finally {
+                    callAction();
+                }
+            }
+
+            void callAction() {
+                try {
                     action.call();
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    RxJavaPlugins.getInstance().getErrorHandler().handleError(ex);
                 }
             }
         };

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

@@ -15,18 +15,14 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.*;
 
-import rx.Observable;
-import rx.Observer;
+import rx.*;
 import rx.functions.Action0;
+import rx.observers.TestSubscriber;
 
 public class OperatorDoAfterTerminateTest {
 
@@ -65,4 +61,37 @@ public void nullActionShouldBeCheckedInConstructor() {
             assertEquals("Action can not be null", expected.getMessage());
         }
     }
+
+    @Test
+    public void nullFinallyActionShouldBeCheckedASAP() {
+        try {
+            Observable
+                    .just("value")
+                    .doAfterTerminate(null);
+
+            fail();
+        } catch (NullPointerException expected) {
+
+        }
+    }
+
+    @Test
+    public void ifFinallyActionThrowsExceptionShouldNotBeSwallowedAndActionShouldBeCalledOnce() {
+        Action0 finallyAction = mock(Action0.class);
+        doThrow(new IllegalStateException()).when(finallyAction).call();
+
+        TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
+
+        Observable
+                .just("value")
+                .doAfterTerminate(finallyAction)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertValue("value");
+
+        verify(finallyAction).call();
+        // Actual result:
+        // Not only IllegalStateException was swallowed
+        // But finallyAction was called twice!
+    }
 }