Merge pull request #3377 from akarnokd/TakeReentrancyFix2x

2.x: fix take() reentrancy problem.

Author: abersnaze

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();
                 }
             }

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();
+    }
 }

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();
+    }
 }