Fix Exception Handling from SafeObserver and Subscriptions

Author: benjchristensen

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

@@ -148,7 +148,7 @@ protected void _onError(Throwable e) {
                 throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
             }
         }
-        // if we did not throw about we will unsubscribe here, if onError failed then unsubscribe happens in the catch
+        // if we did not throw above we will unsubscribe here, if onError failed then unsubscribe happens in the catch
         try {
             unsubscribe();
         } catch (RuntimeException unsubscribeException) {

rxjava-core/src/main/java/rx/subscriptions/CompositeSubscription.java

@@ -18,6 +18,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Subscription;
@@ -141,7 +142,7 @@ public void unsubscribe() {
     }
 
     private static void unsubscribeFromAll(Subscription[] subscriptions) {
-        final Collection<Throwable> es = new ArrayList<Throwable>();
+        final List<Throwable> es = new ArrayList<Throwable>();
         for (Subscription s : subscriptions) {
             try {
                 s.unsubscribe();
@@ -150,8 +151,18 @@ private static void unsubscribeFromAll(Subscription[] subscriptions) {
             }
         }
         if (!es.isEmpty()) {
-            throw new CompositeException(
-                    "Failed to unsubscribe to 1 or more subscriptions.", es);
+            if (es.size() == 1) {
+                Throwable t = es.get(0);
+                if (t instanceof RuntimeException) {
+                    throw (RuntimeException) t;
+                } else {
+                    throw new CompositeException(
+                            "Failed to unsubscribe to 1 or more subscriptions.", es);
+                }
+            } else {
+                throw new CompositeException(
+                        "Failed to unsubscribe to 2 or more subscriptions.", es);
+            }
         }
     }
 }

rxjava-core/src/test/java/rx/subscriptions/CompositeSubscriptionTest.java

@@ -116,12 +116,52 @@ public void unsubscribe() {
             }
         });
 
+        try {
+            s.unsubscribe();
+            fail("Expecting an exception");
+        } catch (RuntimeException e) {
+            // we expect this
+            assertEquals(e.getMessage(), "failed on first one");
+        }
+
+        // we should still have unsubscribed to the second one
+        assertEquals(1, counter.get());
+    }
+
+    @Test
+    public void testCompositeException() {
+        final AtomicInteger counter = new AtomicInteger();
+        CompositeSubscription s = new CompositeSubscription();
+        s.add(new Subscription() {
+
+            @Override
+            public void unsubscribe() {
+                throw new RuntimeException("failed on first one");
+            }
+        });
+
+        s.add(new Subscription() {
+
+            @Override
+            public void unsubscribe() {
+                throw new RuntimeException("failed on second one too");
+            }
+        });
+
+        s.add(new Subscription() {
+
+            @Override
+            public void unsubscribe() {
+                counter.incrementAndGet();
+            }
+        });
+
         try {
             s.unsubscribe();
             fail("Expecting an exception");
         } catch (CompositeException e) {
             // we expect this
-            assertEquals(1, e.getExceptions().size());
+            assertEquals(e.getExceptions().size(), 2);
         }
 
         // we should still have unsubscribed to the second one