Fix for take() reentrancy bug.

Author: akarnokd

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;

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