implement backpressure for OperatorAll and include last value in cause if exception occurs

Author: davidmoten

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

@@ -17,7 +17,10 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Func1;
+import rx.internal.producers.SingleDelayedProducer;
 
 /**
  * Returns an Observable that emits a Boolean that indicates whether all items emitted by an
@@ -34,16 +37,23 @@ public OperatorAll(Func1<? super T, Boolean> predicate) {
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Boolean> child) {
+        final SingleDelayedProducer<Boolean> producer = new SingleDelayedProducer<Boolean>(child);
         Subscriber<T> s = new Subscriber<T>() {
             boolean done;
 
             @Override
             public void onNext(T t) {
-                boolean result = predicate.call(t);
+                Boolean result;
+                try {
+                    result = predicate.call(t);
+                } catch (Throwable e) {
+                    Exceptions.throwIfFatal(e);
+                    onError(OnErrorThrowable.addValueAsLastCause(e, t));
+                    return;
+                }
                 if (!result && !done) {
                     done = true;
-                    child.onNext(false);
-                    child.onCompleted();
+                    producer.setValue(false);
                     unsubscribe();
                 } else {
                 	// if we drop values we must replace them upstream as downstream won't receive and request more
@@ -60,12 +70,12 @@ public void onError(Throwable e) {
             public void onCompleted() {
                 if (!done) {
                     done = true;
-                    child.onNext(true);
-                    child.onCompleted();
+                    producer.setValue(true);
                 }
             }
         };
         child.add(s);
+        child.setProducer(producer);
         return s;
     }
 }

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

@@ -19,12 +19,16 @@
 import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.math3.analysis.FunctionUtils;
 import org.junit.Test;
 
 import rx.*;
+import rx.Observable.OnSubscribe;
 import rx.functions.Func1;
+import rx.observers.TestSubscriber;
 
 public class OperatorAllTest {
 
@@ -128,4 +132,52 @@ public Observable<Integer> call(Boolean t1) {
         });
         assertEquals((Object)2, source.toBlocking().first());
     }
+    
+    @Test
+    public void testBackpressureIfNoneRequestedNoneShouldBeDelivered() {
+        TestSubscriber<Boolean> ts = new TestSubscriber<Boolean>(0);
+        Observable.empty().all(new Func1<Object, Boolean>() {
+            @Override
+            public Boolean call(Object t1) {
+                return false;
+            }
+        }).subscribe(ts);
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testBackpressureIfOneRequestedOneShouldBeDelivered() {
+        TestSubscriber<Boolean> ts = new TestSubscriber<Boolean>(1);
+        Observable.empty().all(new Func1<Object, Boolean>() {
+            @Override
+            public Boolean call(Object object) {
+                return false;
+            }
+        }).subscribe(ts);
+        ts.assertTerminalEvent();
+        ts.assertNoErrors();
+        ts.assertCompleted();
+        ts.assertValue(true);
+    }
+    
+    @Test
+    public void testPredicateThrowsExceptionAndValueInCauseMessage() {
+        TestSubscriber<Boolean> ts = new TestSubscriber<Boolean>(0);
+        final IllegalArgumentException ex = new IllegalArgumentException();
+        Observable.just("Boo!").all(new Func1<Object, Boolean>() {
+            @Override
+            public Boolean call(Object object) {
+                throw ex;
+            }
+        }).subscribe(ts);
+        ts.assertTerminalEvent();
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+        List<Throwable> errors = ts.getOnErrorEvents();
+        assertEquals(1, errors.size());
+        assertEquals(ex, errors.get(0));
+        assertTrue(ex.getCause().getMessage().contains("Boo!"));
+    }
 }