OperatorAll - prevent multiple terminal events

Author: davidmoten

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

@@ -61,7 +61,10 @@ public void onNext(T t) {
 
             @Override
             public void onError(Throwable e) {
-                child.onError(e);
+                if (!done) {
+                    done = true;
+                    child.onError(e);
+                }
             }
 
             @Override

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

@@ -25,6 +25,7 @@
 import org.junit.Test;
 
 import rx.*;
+import rx.Observable.OnSubscribe;
 import rx.functions.Func1;
 import rx.observers.TestSubscriber;
 
@@ -178,4 +179,34 @@ public Boolean call(Object object) {
         assertEquals(ex, errors.get(0));
         assertTrue(ex.getCause().getMessage().contains("Boo!"));
     }
+    
+    @Test
+    public void testDoesNotEmitMultipleTerminalEvents() {
+        TestSubscriber<Boolean> ts = TestSubscriber.create();
+        Observable.create(new OnSubscribe<Integer>() {
+
+            @Override
+            public void call(final Subscriber<? super Integer> sub) {
+                sub.setProducer(new Producer() {
+
+                    @Override
+                    public void request(long n) {
+                        if (n > 0) {
+                            sub.onNext(1);
+                            sub.onCompleted();
+                        }
+                    }
+                });
+            }
+        })
+        .all(new Func1<Integer,Boolean>() {
+
+            @Override
+            public Boolean call(Integer t) {
+                throw new RuntimeException("boo");
+            }})
+        .unsafeSubscribe(ts);
+        ts.assertError(RuntimeException.class);
+        ts.assertNotCompleted();
+    }
 }