Take: Fix Terminal State Handling

- even though unit tests don’t see it because of SafeObserver, the Take operator should not emit onCompleted more than once

Author: benjchristensen

rxjava-core/src/main/java/rx/operators/OperatorTake.java

@@ -47,22 +47,28 @@ public Observer<T> call(final Observer<? super T> o, final OperatorSubscription
         return new Observer<T>() {
 
             int count = 0;
+            boolean completed = false;
 
             @Override
             public void onCompleted() {
-                o.onCompleted();
+                if (!completed) {
+                    o.onCompleted();
+                }
             }
 
             @Override
             public void onError(Throwable e) {
-                o.onError(e);
+                if (!completed) {
+                    o.onError(e);
+                }
             }
 
             @Override
             public void onNext(T i) {
                 if (!s.isUnsubscribed()) {
                     o.onNext(i);
                     if (++count >= limit) {
+                        completed = true;
                         o.onCompleted();
                         s.unsubscribe();
                     }

rxjava-core/src/perf/java/rx/operators/OperatorTakePerformance.java

@@ -27,11 +27,11 @@ public void call() {
     /**
      * Observable.range(0, 10).take(5);
      * 
-     * Run: 10 - 6,660,042 ops/sec
-     * Run: 11 - 6,721,423 ops/sec
-     * Run: 12 - 6,556,035 ops/sec
-     * Run: 13 - 6,692,284 ops/sec
-     * Run: 14 - 6,731,287 ops/sec
+     * Run: 10 - 8,780,556 ops/sec
+     * Run: 11 - 8,822,590 ops/sec
+     * Run: 12 - 8,842,733 ops/sec
+     * Run: 13 - 8,825,486 ops/sec
+     * Run: 14 - 8,771,545 ops/sec
      */
     public long timeTake5() {
 

rxjava-core/src/test/java/rx/operators/OperatorTakeTest.java

@@ -18,22 +18,19 @@
 import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
-import static rx.operators.OperatorTake.*;
 
 import java.util.Arrays;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 import org.mockito.InOrder;
 
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
-import rx.operators.OperationSkipTest.CustomException;
-import rx.schedulers.TestScheduler;
-import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
+import rx.util.functions.Action0;
 import rx.util.functions.Func1;
 
 public class OperatorTakeTest {