add backpressure support for OperatorAny and include last value in exception cause

Author: davidmoten

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

@@ -19,7 +19,10 @@
 import rx.Observable;
 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 {@link Observable} that emits <code>true</code> if any element of
@@ -36,23 +39,29 @@ public OperatorAny(Func1<? super T, Boolean> predicate, boolean returnOnEmpty) {
 
     @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 hasElements;
             boolean done;
 
             @Override
             public void onNext(T t) {
                 hasElements = true;
-                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(!returnOnEmpty);
-                    child.onCompleted();
+                    producer.setValue(!returnOnEmpty);
                     unsubscribe();
-                } else {
-                    // if we drop values we must replace them upstream as downstream won't receive and request more
-                    request(1);
-                }
+                } 
+                // note that don't need to request more of upstream because this subscriber 
+                // defaults to requesting Long.MAX_VALUE
             }
 
             @Override
@@ -65,16 +74,16 @@ public void onCompleted() {
                 if (!done) {
                     done = true;
                     if (hasElements) {
-                        child.onNext(false);
+                        producer.setValue(false);
                     } else {
-                        child.onNext(returnOnEmpty);
+                        producer.setValue(returnOnEmpty);
                     }
-                    child.onCompleted();
                 }
             }
 
         };
         child.add(s);
+        child.setProducer(producer);
         return s;
     }
 }

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

@@ -19,13 +19,15 @@
 import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
 import rx.*;
 import rx.functions.Func1;
 import rx.internal.util.UtilityFunctions;
+import rx.observers.TestSubscriber;
 
 public class OperatorAnyTest {
 
@@ -220,4 +222,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.just(1).exists(new Func1<Object, Boolean>() {
+            @Override
+            public Boolean call(Object t1) {
+                return true;
+            }
+        }).subscribe(ts);
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testBackpressureIfOneRequestedOneShouldBeDelivered() {
+        TestSubscriber<Boolean> ts = new TestSubscriber<Boolean>(1);
+        Observable.just(1).exists(new Func1<Object, Boolean>() {
+            @Override
+            public Boolean call(Object object) {
+                return true;
+            }
+        }).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!").exists(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!"));
+    }
 }