Re-implemented Take Operator with Bind

- also simplified implementation to not worry about thread-safety as per Rx contract
- performance improvement from 4,033,468 ops/sec -> 6,731,287 ops/sec

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -88,7 +88,7 @@
 import rx.operators.OperationSum;
 import rx.operators.OperationSwitch;
 import rx.operators.OperationSynchronize;
-import rx.operators.OperationTake;
+import rx.operators.OperatorTake;
 import rx.operators.OperationTakeLast;
 import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
@@ -106,6 +106,7 @@
 import rx.operators.OperationUsing;
 import rx.operators.OperationWindow;
 import rx.operators.OperationZip;
+import rx.operators.OperatorTakeTimed;
 import rx.operators.SafeObservableSubscription;
 import rx.operators.SafeObserver;
 import rx.plugins.RxJavaObservableExecutionHook;
@@ -7384,7 +7385,7 @@ public final Observable<T> synchronize(Object lock) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#take">RxJava Wiki: take()</a>
      */
     public final Observable<T> take(final int num) {
-        return create(OperationTake.take(this, num));
+        return bind(new OperatorTake<T>(num));
     }
 
     /**
@@ -7422,7 +7423,7 @@ public final Observable<T> take(long time, TimeUnit unit) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#take">RxJava Wiki: take()</a>
      */
     public final Observable<T> take(long time, TimeUnit unit, Scheduler scheduler) {
-        return create(new OperationTake.TakeTimed<T>(this, time, unit, scheduler));
+        return create(new OperatorTakeTimed.TakeTimed<T>(this, time, unit, scheduler));
     }
 
     /**

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

@@ -0,0 +1,75 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import rx.Observable.OperatorSubscription;
+import rx.Observer;
+import rx.util.functions.Func2;
+
+/**
+ * Returns an Observable that emits the first <code>num</code> items emitted by the source
+ * Observable.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/take.png">
+ * <p>
+ * You can choose to pay attention only to the first <code>num</code> items emitted by an
+ * Observable by using the take operation. This operation returns an Observable that will invoke a
+ * subscribing Observer's <code>onNext</code> function a maximum of <code>num</code> times before
+ * invoking <code>onCompleted</code>.
+ */
+public final class OperatorTake<T> implements Func2<Observer<? super T>, OperatorSubscription, Observer<? super T>> {
+
+    final int limit;
+
+    public OperatorTake(int limit) {
+        this.limit = limit;
+    }
+
+    @Override
+    public Observer<T> call(final Observer<? super T> o, final OperatorSubscription s) {
+        if (limit == 0) {
+            o.onCompleted();
+            s.unsubscribe();
+        }
+        return new Observer<T>() {
+
+            int count = 0;
+
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+
+            @Override
+            public void onNext(T i) {
+                if (!s.isUnsubscribed()) {
+                    o.onNext(i);
+                    if (++count >= limit) {
+                        o.onCompleted();
+                        s.unsubscribe();
+                    }
+                }
+            }
+
+        };
+    }
+
+}

rxjava-core/src/main/java/rx/operators/OperationTake.java renamed to rxjava-core/src/main/java/rx/operators/OperatorTakeTimed.java

@@ -38,7 +38,7 @@
  * subscribing Observer's <code>onNext</code> function a maximum of <code>num</code> times before
  * invoking <code>onCompleted</code>.
  */
-public final class OperationTake {
+public final class OperatorTakeTimed {
 
     /**
      * Returns a specified number of contiguous values from the start of an observable sequence.

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

@@ -27,11 +27,11 @@ public void call() {
     /**
      * Observable.range(0, 10).take(5);
      * 
-     * Run: 10 - 3,951,557 ops/sec
-     * Run: 11 - 3,981,329 ops/sec
-     * Run: 12 - 3,988,949 ops/sec
-     * Run: 13 - 3,925,971 ops/sec
-     * Run: 14 - 4,033,468 ops/sec
+     * 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
      */
     public long timeTake5() {
 

rxjava-core/src/test/java/rx/operators/OperationTakeTest.java renamed to rxjava-core/src/test/java/rx/operators/OperatorTakeTest.java

@@ -18,8 +18,9 @@
 import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
-import static rx.operators.OperationTake.*;
+import static rx.operators.OperatorTake.*;
 
+import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -35,12 +36,12 @@
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func1;
 
-public class OperationTakeTest {
+public class OperatorTakeTest {
 
     @Test
     public void testTake1() {
-        Observable<String> w = Observable.from("one", "two", "three");
-        Observable<String> take = Observable.create(take(w, 2));
+        Observable<String> w = Observable.from(Arrays.asList("one", "two", "three"));
+        Observable<String> take = w.bind(new OperatorTake<String>(2));
 
         @SuppressWarnings("unchecked")
         Observer<String> aObserver = mock(Observer.class);
@@ -54,8 +55,8 @@ public void testTake1() {
 
     @Test
     public void testTake2() {
-        Observable<String> w = Observable.from("one", "two", "three");
-        Observable<String> take = Observable.create(take(w, 1));
+        Observable<String> w = Observable.from(Arrays.asList("one", "two", "three"));
+        Observable<String> take = w.bind(new OperatorTake<String>(1));
 
         @SuppressWarnings("unchecked")
         Observer<String> aObserver = mock(Observer.class);
@@ -69,7 +70,7 @@ public void testTake2() {
 
     @Test(expected = IllegalArgumentException.class)
     public void testTakeWithError() {
-        Observable.from(1, 2, 3).take(1).map(new Func1<Integer, Integer>() {
+        Observable.from(Arrays.asList(1, 2, 3)).take(1).map(new Func1<Integer, Integer>() {
             public Integer call(Integer t1) {
                 throw new IllegalArgumentException("some error");
             }
@@ -78,7 +79,7 @@ public Integer call(Integer t1) {
 
     @Test
     public void testTakeWithErrorHappeningInOnNext() {
-        Observable<Integer> w = Observable.from(1, 2, 3).take(2).map(new Func1<Integer, Integer>() {
+        Observable<Integer> w = Observable.from(Arrays.asList(1, 2, 3)).take(2).map(new Func1<Integer, Integer>() {
             public Integer call(Integer t1) {
                 throw new IllegalArgumentException("some error");
             }
@@ -94,7 +95,7 @@ public Integer call(Integer t1) {
 
     @Test
     public void testTakeWithErrorHappeningInTheLastOnNext() {
-        Observable<Integer> w = Observable.from(1, 2, 3).take(1).map(new Func1<Integer, Integer>() {
+        Observable<Integer> w = Observable.from(Arrays.asList(1, 2, 3)).take(1).map(new Func1<Integer, Integer>() {
             public Integer call(Integer t1) {
                 throw new IllegalArgumentException("some error");
             }
@@ -122,7 +123,7 @@ public Subscription onSubscribe(Observer<? super String> observer) {
         @SuppressWarnings("unchecked")
         Observer<String> aObserver = mock(Observer.class);
 
-        Observable.create(take(source, 1)).subscribe(aObserver);
+        source.bind(new OperatorTake<String>(1)).subscribe(aObserver);
 
         verify(aObserver, times(1)).onNext("one");
         // even though onError is called we take(1) so shouldn't see it
@@ -152,7 +153,7 @@ public void unsubscribe() {
         @SuppressWarnings("unchecked")
         Observer<String> aObserver = mock(Observer.class);
 
-        Observable.create(take(source, 0)).subscribe(aObserver);
+        source.bind(new OperatorTake<String>(0)).subscribe(aObserver);
         assertTrue("source subscribed", subscribed.get());
         assertTrue("source unsubscribed", unSubscribed.get());
 
@@ -171,7 +172,7 @@ public void testUnsubscribeAfterTake() {
 
         @SuppressWarnings("unchecked")
         Observer<String> aObserver = mock(Observer.class);
-        Observable<String> take = Observable.create(take(w, 1));
+        Observable<String> take = w.bind(new OperatorTake<String>(1));
         take.subscribe(aObserver);
 
         // wait for the Observable to complete
@@ -228,99 +229,4 @@ public void run() {
             return s;
         }
     }
-
-    @Test
-    public void testTakeTimed() {
-        TestScheduler scheduler = new TestScheduler();
-
-        PublishSubject<Integer> source = PublishSubject.create();
-
-        Observable<Integer> result = source.take(1, TimeUnit.SECONDS, scheduler);
-
-        Observer<Object> o = mock(Observer.class);
-
-        result.subscribe(o);
-
-        source.onNext(1);
-        source.onNext(2);
-        source.onNext(3);
-
-        scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
-
-        source.onNext(4);
-
-        InOrder inOrder = inOrder(o);
-        inOrder.verify(o).onNext(1);
-        inOrder.verify(o).onNext(2);
-        inOrder.verify(o).onNext(3);
-        inOrder.verify(o).onCompleted();
-        inOrder.verifyNoMoreInteractions();
-
-        verify(o, never()).onNext(4);
-        verify(o, never()).onError(any(Throwable.class));
-    }
-
-    @Test
-    public void testTakeTimedErrorBeforeTime() {
-        TestScheduler scheduler = new TestScheduler();
-
-        PublishSubject<Integer> source = PublishSubject.create();
-
-        Observable<Integer> result = source.take(1, TimeUnit.SECONDS, scheduler);
-
-        Observer<Object> o = mock(Observer.class);
-
-        result.subscribe(o);
-
-        source.onNext(1);
-        source.onNext(2);
-        source.onNext(3);
-        source.onError(new CustomException());
-
-        scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
-
-        source.onNext(4);
-
-        InOrder inOrder = inOrder(o);
-        inOrder.verify(o).onNext(1);
-        inOrder.verify(o).onNext(2);
-        inOrder.verify(o).onNext(3);
-        inOrder.verify(o).onError(any(CustomException.class));
-        inOrder.verifyNoMoreInteractions();
-
-        verify(o, never()).onCompleted();
-        verify(o, never()).onNext(4);
-    }
-
-    @Test
-    public void testTakeTimedErrorAfterTime() {
-        TestScheduler scheduler = new TestScheduler();
-
-        PublishSubject<Integer> source = PublishSubject.create();
-
-        Observable<Integer> result = source.take(1, TimeUnit.SECONDS, scheduler);
-
-        Observer<Object> o = mock(Observer.class);
-
-        result.subscribe(o);
-
-        source.onNext(1);
-        source.onNext(2);
-        source.onNext(3);
-
-        scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
-
-        source.onNext(4);
-        source.onError(new CustomException());
-
-        InOrder inOrder = inOrder(o);
-        inOrder.verify(o).onNext(1);
-        inOrder.verify(o).onNext(2);
-        inOrder.verify(o).onNext(3);
-        inOrder.verify(o).onCompleted();
-        inOrder.verifyNoMoreInteractions();
-
-        verify(o, never()).onNext(4);
-        verify(o, never()).onError(any(CustomException.class));
-    }
 }