Operator TakeTimed

Author: akarnokd

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

@@ -68,7 +68,6 @@
 import rx.operators.OperationSkipUntil;
 import rx.operators.OperationSwitch;
 import rx.operators.OperationTakeLast;
-import rx.operators.OperationTakeTimed;
 import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
 import rx.operators.OperationThrottleFirst;
@@ -121,6 +120,7 @@
 import rx.operators.OperatorSkipWhile;
 import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorTake;
+import rx.operators.OperatorTakeTimed;
 import rx.operators.OperatorTimeout;
 import rx.operators.OperatorTimeoutWithSelector;
 import rx.operators.OperatorTimestamp;
@@ -6447,7 +6447,7 @@ public final Observable<T> take(long time, TimeUnit unit) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-take">RxJava Wiki: take()</a>
      */
     public final Observable<T> take(long time, TimeUnit unit, Scheduler scheduler) {
-        return create(new OperationTakeTimed.TakeTimed<T>(this, time, unit, scheduler));
+        return lift(new OperatorTakeTimed<T>(time, unit, scheduler));
     }
 
     /**

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

@@ -0,0 +1,84 @@
+/**
+ * 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 java.util.concurrent.TimeUnit;
+import rx.Observable.Operator;
+import rx.Scheduler;
+import rx.Scheduler.Worker;
+import rx.Subscriber;
+import rx.functions.Action0;
+import rx.observers.SerializedSubscriber;
+
+/**
+ * Takes values from the source until the specific time ellapses.
+ * 
+ * @param <T>
+ *            the result value type
+ */
+public final class OperatorTakeTimed<T> implements Operator<T, T> {
+    final long time;
+    final TimeUnit unit;
+    final Scheduler scheduler;
+
+    public OperatorTakeTimed(long time, TimeUnit unit, Scheduler scheduler) {
+        this.time = time;
+        this.unit = unit;
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Subscriber<? super T> call(Subscriber<? super T> child) {
+        Worker worker = scheduler.createWorker();
+        child.add(worker);
+        
+        TakeSubscriber<T> ts = new TakeSubscriber<T>(new SerializedSubscriber<T>(child));
+        worker.schedule(ts, time, unit);
+        return ts;
+    }
+    /** Subscribed to source and scheduled on a worker. */
+    static final class TakeSubscriber<T> extends Subscriber<T> implements Action0 {
+        final Subscriber<? super T> child;
+        public TakeSubscriber(Subscriber<? super T> child) {
+            super(child);
+            this.child = child;
+        }
+
+        @Override
+        public void onNext(T t) {
+            child.onNext(t);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            child.onError(e);
+            unsubscribe();
+        }
+
+        @Override
+        public void onCompleted() {
+            child.onCompleted();
+            unsubscribe();
+        }
+
+        @Override
+        public void call() {
+            onCompleted();
+        }
+        
+        
+    }
+}