OperatorAll - prevent multiple terminal events (#4244)

Author: davidmoten

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

@@ -20,6 +20,7 @@
 import rx.exceptions.Exceptions;
 import rx.functions.Func1;
 import rx.internal.producers.SingleDelayedProducer;
+import rx.plugins.RxJavaHooks;
 
 /**
  * Returns an Observable that emits a Boolean that indicates whether all items emitted by an
@@ -43,14 +44,17 @@ public Subscriber<? super T> call(final Subscriber<? super Boolean> child) {
 
             @Override
             public void onNext(T t) {
+                if (done) {
+                    return;
+                }
                 Boolean result;
                 try {
                     result = predicate.call(t);
                 } catch (Throwable e) {
                     Exceptions.throwOrReport(e, this, t);
                     return;
                 }
-                if (!result && !done) {
+                if (!result) {
                     done = true;
                     producer.setValue(false);
                     unsubscribe();
@@ -61,7 +65,12 @@ public void onNext(T t) {
 
             @Override
             public void onError(Throwable e) {
-                child.onError(e);
+                if (!done) {
+                    done = true;
+                    child.onError(e);
+                } else {
+                    RxJavaHooks.onError(e);
+                }
             }
 
             @Override

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

@@ -20,13 +20,17 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
 import rx.*;
+import rx.Observable.OnSubscribe;
+import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.observers.TestSubscriber;
+import rx.plugins.RxJavaHooks;
 
 public class OperatorAllTest {
 
@@ -178,4 +182,114 @@ 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();
+    }
+    
+    @Test
+    public void testUpstreamEmitsOnNextAfterFailureWithoutCheckingSubscription() {
+        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 > 1) {
+                            sub.onNext(1);
+                            sub.onNext(2);
+                        }
+                    }
+                });
+            }
+        })
+        .all(new Func1<Integer,Boolean>() {
+            boolean once = true;
+            @Override
+            public Boolean call(Integer t) {
+                if (once)
+                    throw new RuntimeException("boo");
+                else  {
+                    once = false;
+                    return true; 
+                }
+            }})
+        .unsafeSubscribe(ts);
+        ts.assertNoValues();
+        ts.assertError(RuntimeException.class);
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testDoesNotEmitMultipleErrorEventsAndReportsSecondErrorToHooks() {
+        try {
+            final List<Throwable> list = new CopyOnWriteArrayList<Throwable>();
+            RxJavaHooks.setOnError(new Action1<Throwable>() {
+
+                @Override
+                public void call(Throwable t) {
+                    list.add(t);
+                }
+            });
+            TestSubscriber<Boolean> ts = TestSubscriber.create();
+            final RuntimeException e1 = new RuntimeException();
+            final Throwable e2 = new RuntimeException();
+            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.onError(e2);
+                            }
+                        }
+                    });
+                }
+            }).all(new Func1<Integer, Boolean>() {
+
+                @Override
+                public Boolean call(Integer t) {
+                    throw e1;
+                }
+            }).unsafeSubscribe(ts);
+            ts.assertNotCompleted();
+            assertEquals(Arrays.asList(e1), ts.getOnErrorEvents());
+            assertEquals(Arrays.asList(e2), list);
+        } finally {
+            RxJavaHooks.setOnError(null);
+        }
+    }
 }