Merge pull request ReactiveX#3377 from akarnokd/TakeReentrancyFix2x

abersnaze · abersnaze · commit 9bc355f1c2dc · 2015-09-28T16:52:36.000-07:00
2.x: fix take() reentrancy problem.
diff --git a/src/main/java/io/reactivex/internal/operators/OperatorTake.java b/src/main/java/io/reactivex/internal/operators/OperatorTake.java
@@ -62,9 +62,10 @@ public void onSubscribe(Subscription s) {
         }
         @Override
         public void onNext(T t) {
-            if (!done) {
+            if (!done && remaining-- > 0) {
+                boolean stop = remaining == 0;
                 actual.onNext(t);
-                if (--remaining == 0L) {
+                if (stop) {
                     onComplete();
                 }
             }
diff --git a/src/test/java/io/reactivex/internal/operators/OperatorTakeTest.java b/src/test/java/io/reactivex/internal/operators/OperatorTakeTest.java
@@ -30,6 +30,7 @@
 import io.reactivex.exceptions.TestException;
 import io.reactivex.internal.subscriptions.*;
 import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subjects.PublishSubject;
 import io.reactivex.subscribers.TestSubscriber;
 
 public class OperatorTakeTest {
@@ -420,4 +421,19 @@ public void onNext(Integer t) {
         ts.assertError(TestException.class);
         ts.assertNotComplete();
     }
+    
+    @Test
+    public void testReentrantTake() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        
+        TestSubscriber<Integer> ts = new TestSubscriber<>();
+        
+        source.take(1).doOnNext(v -> source.onNext(2)).subscribe(ts);
+        
+        source.onNext(1);
+        
+        ts.assertValue(1);
+        ts.assertNoErrors();
+        ts.assertComplete();
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/nbp/NbpOperatorTakeTest.java b/src/test/java/io/reactivex/internal/operators/nbp/NbpOperatorTakeTest.java
@@ -31,6 +31,7 @@
 import io.reactivex.exceptions.TestException;
 import io.reactivex.internal.disposables.EmptyDisposable;
 import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subjects.nbp.NbpPublishSubject;
 import io.reactivex.subscribers.nbp.NbpTestSubscriber;
 
 public class NbpOperatorTakeTest {
@@ -338,4 +339,19 @@ public void onNext(Integer t) {
         ts.assertError(TestException.class);
         ts.assertNotComplete();
     }
+    
+    @Test
+    public void testReentrantTake() {
+        NbpPublishSubject<Integer> source = NbpPublishSubject.create();
+        
+        NbpTestSubscriber<Integer> ts = new NbpTestSubscriber<>();
+        
+        source.take(1).doOnNext(v -> source.onNext(2)).subscribe(ts);
+        
+        source.onNext(1);
+        
+        ts.assertValue(1);
+        ts.assertNoErrors();
+        ts.assertComplete();
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -62,9 +62,10 @@ public void onSubscribe(Subscription s) {`
`62`	`62`	`}`
`63`	`63`	`@Override`
`64`	`64`	`public void onNext(T t) {`
`65`		`- if (!done) {`
	`65`	`+ if (!done && remaining-- > 0) {`
	`66`	`+ boolean stop = remaining == 0;`
`66`	`67`	`actual.onNext(t);`
`67`		`- if (--remaining == 0L) {`
	`68`	`+ if (stop) {`
`68`	`69`	`onComplete();`
`69`	`70`	`}`
`70`	`71`	`}`