From a10d8dca55c0e22a319ae9f6c2e5a5060e9e0fcc Mon Sep 17 00:00:00 2001
From: akarnokd <akarnokd@gmail.com>
Date: Mon, 26 Mar 2018 14:44:49 +0200
Subject: [PATCH] 1.x: Fix take() to route late errors to RxJavaHooks

---
 .../rx/internal/operators/OperatorTake.java   |  3 ++
 .../internal/operators/OperatorTakeTest.java  | 40 ++++++++++++++++++-
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/src/main/java/rx/internal/operators/OperatorTake.java b/src/main/java/rx/internal/operators/OperatorTake.java
index 0be75b4f2c..d49f155d9d 100644
--- a/src/main/java/rx/internal/operators/OperatorTake.java
+++ b/src/main/java/rx/internal/operators/OperatorTake.java
@@ -19,6 +19,7 @@
 
 import rx.*;
 import rx.Observable.Operator;
+import rx.plugins.RxJavaHooks;
 
 /**
  * An {@code Observable} that emits the first {@code num} items emitted by the source {@code Observable}.
@@ -66,6 +67,8 @@ public void onError(Throwable e) {
                     } finally {
                         unsubscribe();
                     }
+                } else {
+                    RxJavaHooks.onError(e);
                 }
             }
 
diff --git a/src/test/java/rx/internal/operators/OperatorTakeTest.java b/src/test/java/rx/internal/operators/OperatorTakeTest.java
index edf17bac10..0885a8b343 100644
--- a/src/test/java/rx/internal/operators/OperatorTakeTest.java
+++ b/src/test/java/rx/internal/operators/OperatorTakeTest.java
@@ -19,7 +19,7 @@
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
-import java.util.Arrays;
+import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 
@@ -27,10 +27,13 @@
 import org.mockito.InOrder;
 
 import rx.*;
+import rx.Observable;
 import rx.Observable.OnSubscribe;
+import rx.Observer;
 import rx.exceptions.TestException;
 import rx.functions.*;
 import rx.observers.*;
+import rx.plugins.RxJavaHooks;
 import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
 
@@ -457,4 +460,39 @@ public void takeZero() {
         ts.assertCompleted();
     }
 
+    @Test
+    public void crashReportedToHooks() {
+        final List<Throwable> errors = Collections.synchronizedList(new ArrayList<Throwable>());
+        RxJavaHooks.setOnError(new Action1<Throwable>() {
+            @Override
+            public void call(Throwable error) {
+                errors.add(error);
+            }
+        });
+
+        try {
+            Observable.just("1")
+                .take(1)
+                .toSingle()
+                .subscribe(
+                        new Action1<String>() {
+                            @Override
+                            public void call(String it) {
+                                throw new TestException("bla");
+                            }
+                        },
+                        new Action1<Throwable>() {
+                            @Override
+                            public void call(Throwable error) {
+                                errors.add(new AssertionError());
+                            }
+                        }
+                );
+            
+            assertEquals("" + errors, 1, errors.size());
+            assertTrue("" + errors.get(0), errors.get(0).getMessage().equals("bla"));
+        } finally {
+            RxJavaHooks.setOnError(null);
+        }
+    }
 }