Merge pull request #2993 from davidmoten/take-while-cause

takeWhile(predicate) - include last value in error cause

Author: akarnokd

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

@@ -17,6 +17,8 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Func1;
 import rx.functions.Func2;
 
@@ -52,18 +54,19 @@ public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
             private boolean done = false;
 
             @Override
-            public void onNext(T args) {
+            public void onNext(T t) {
                 boolean isSelected;
                 try {
-                    isSelected = predicate.call(args, counter++);
+                    isSelected = predicate.call(t, counter++);
                 } catch (Throwable e) {
                     done = true;
-                    subscriber.onError(e);
+                    Exceptions.throwIfFatal(e);
+                    subscriber.onError(OnErrorThrowable.addValueAsLastCause(e, t));
                     unsubscribe();
                     return;
                 }
                 if (isSelected) {
-                    subscriber.onNext(args);
+                    subscriber.onNext(t);
                 } else {
                     done = true;
                     subscriber.onCompleted();

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

@@ -15,6 +15,7 @@
  */
 package rx.internal.operators;
 
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
@@ -25,6 +26,7 @@
 
 import rx.*;
 import rx.Observable.OnSubscribe;
+import rx.exceptions.TestException;
 import rx.functions.Func1;
 import rx.observers.TestSubscriber;
 import rx.subjects.*;
@@ -261,4 +263,20 @@ public Boolean call(Integer t1) {
 
         Assert.assertFalse("Unsubscribed!", ts.isUnsubscribed());
     }
+    
+    @Test
+    public void testErrorCauseIncludesLastValue() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable.just("abc").takeWhile(new Func1<String, Boolean>() {
+            @Override
+            public Boolean call(String t1) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertTerminalEvent();
+        ts.assertNoValues();
+        assertTrue(ts.getOnErrorEvents().get(0).getCause().getMessage().contains("abc"));
+    }
+    
 }