Merge pull request #3384 from akarnokd/TakeReentrancyFix1x

Fix for take() reentrancy bug.
ReactiveX · Sep 30, 2015 · ebe28de · ebe28de
2 parents f2410f8 + 53ff799
commit ebe28de
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 2 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorTake.java b/src/main/java/rx/internal/operators/OperatorTake.java
@@ -68,8 +68,8 @@ public void onError(Throwable e) {
 
             @Override
             public void onNext(T i) {
-                if (!isUnsubscribed()) {
-                    boolean stop = ++count >= limit;
+                if (!isUnsubscribed() && count++ < limit) {
+                    boolean stop = count == limit;
                     child.onNext(i);
                     if (stop && !completed) {
                         completed = true;

diff --git a/src/test/java/rx/internal/operators/OperatorTakeTest.java b/src/test/java/rx/internal/operators/OperatorTakeTest.java
@@ -32,6 +32,7 @@
 import rx.functions.*;
 import rx.observers.*;
 import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
 
 public class OperatorTakeTest {
 
@@ -417,4 +418,24 @@ public void onNext(Integer t) {
         ts.assertError(TestException.class);
         ts.assertNotCompleted();
     }
+
+    @Test
+    public void testReentrantTake() {
+        final PublishSubject<Integer> source = PublishSubject.create();
+
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+        source.take(1).doOnNext(new Action1<Integer>() {
+            @Override
+            public void call(Integer v) {
+                source.onNext(2);
+            }
+        }).subscribe(ts);
+
+        source.onNext(1);
+
+        ts.assertValue(1);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
 }