catch onCompleted unsubscribe error and report to RxJavaPlugin error handler

Author: davidmoten

src/main/java/rx/observers/SafeSubscriber.java

@@ -86,8 +86,16 @@ public void onCompleted() {
                 // handle errors if the onCompleted implementation fails, not just if the Observable fails
                 _onError(e);
             } finally {
-                // auto-unsubscribe
-                unsubscribe();
+                try {
+                    // auto-unsubscribe
+                    unsubscribe();
+                } catch (Throwable e2) {
+                    try {
+                        RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);
+                    } catch (Throwable pluginException) {
+                        handlePluginException(pluginException);
+                    }
+                }
             }
         }
     }

src/test/java/rx/observers/SafeSubscriberTest.java

@@ -15,15 +15,23 @@
  */
 package rx.observers;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import java.lang.reflect.Method;
+import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.*;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
 
-import rx.exceptions.*;
+import rx.exceptions.OnErrorFailedException;
+import rx.exceptions.OnErrorNotImplementedException;
+import rx.exceptions.TestException;
 import rx.functions.Action0;
-import rx.plugins.*;
+import rx.plugins.RxJavaErrorHandler;
+import rx.plugins.RxJavaPlugins;
 import rx.subscriptions.Subscriptions;
 
 public class SafeSubscriberTest {
@@ -227,4 +235,34 @@ public void call() {
 
         safe.onError(new TestException());
     }
+    
+    @Test
+    public void testPluginErrorHandlerReceivesExceptionWhenUnsubscribeAfterCompletionThrows() {
+        final AtomicInteger calls = new AtomicInteger();
+        RxJavaPlugins.getInstance().registerErrorHandler(new RxJavaErrorHandler() {
+            @Override
+            public void handleError(Throwable e) {
+                calls.incrementAndGet();
+            }
+        });
+        
+        final AtomicInteger errors = new AtomicInteger();
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onError(Throwable e) {
+                errors.incrementAndGet();
+            }
+        };
+        SafeSubscriber<Integer> safe = new SafeSubscriber<Integer>(ts);
+        safe.add(Subscriptions.create(new Action0() {
+            @Override
+            public void call() {
+                throw new RuntimeException();
+            }
+        }));
+        
+        safe.onCompleted();
+        assertEquals(1, (int) calls.get());
+        assertEquals(0, (int) errors.get());
+    }
 }