Merge pull request #2991 from davidmoten/take-until-predicate-last-cause

takeUntil(predicate) - include last value in error cause
ReactiveX · May 30, 2015 · 915e2ac · 915e2ac
2 parents da588a2 + 0fbc407
commit 915e2ac
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 4 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorTakeUntilPredicate.java b/src/main/java/rx/internal/operators/OperatorTakeUntilPredicate.java
@@ -18,6 +18,8 @@
 import rx.Observable.Operator;
 import rx.*;
 import rx.annotations.Experimental;
+import rx.exceptions.Exceptions;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Func1;
 
 /**
@@ -37,15 +39,16 @@ private ParentSubscriber(Subscriber<? super T> child) {
         }
 
         @Override
-        public void onNext(T args) {
-            child.onNext(args);
+        public void onNext(T t) {
+            child.onNext(t);
 
             boolean stop = false;
             try {
-                stop = stopPredicate.call(args);
+                stop = stopPredicate.call(t);
             } catch (Throwable e) {
                 done = true;
-                child.onError(e);
+                Exceptions.throwIfFatal(e);
+                child.onError(OnErrorThrowable.addValueAsLastCause(e, t));
                 unsubscribe();
                 return;
             }

diff --git a/src/test/java/rx/internal/operators/OperatorTakeUntilPredicateTest.java b/src/test/java/rx/internal/operators/OperatorTakeUntilPredicateTest.java
@@ -16,6 +16,8 @@
 
 package rx.internal.operators;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
@@ -132,4 +134,20 @@ public void onStart() {
         ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5));
         Assert.assertEquals(0, ts.getOnCompletedEvents().size());
     }
+
+    @Test
+    public void testErrorIncludesLastValueAsCause() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        final TestException e = new TestException("Forced failure");
+        Observable.just("abc").takeUntil(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String t) {
+                throw e;
+            }
+        }).subscribe(ts);
+        ts.assertTerminalEvent();
+        ts.assertNotCompleted();
+        assertEquals(1, (int) ts.getOnErrorEvents().size());
+        assertTrue(ts.getOnErrorEvents().get(0).getCause().getMessage().contains("abc"));
+    }
 }