OperatorSkipTimed

Author: akarnokd

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

@@ -64,7 +64,6 @@
 import rx.operators.OperationReplay;
 import rx.operators.OperationSample;
 import rx.operators.OperationSequenceEqual;
-import rx.operators.OperationSkip;
 import rx.operators.OperationSkipUntil;
 import rx.operators.OperationSwitch;
 import rx.operators.OperationTakeLast;
@@ -118,6 +117,7 @@
 import rx.operators.OperatorSkip;
 import rx.operators.OperatorSkipLast;
 import rx.operators.OperatorSkipLastTimed;
+import rx.operators.OperatorSkipTimed;
 import rx.operators.OperatorSkipWhile;
 import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorTake;
@@ -5547,7 +5547,7 @@ public final Observable<T> skip(long time, TimeUnit unit) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-skip">RxJava Wiki: skip()</a>
      */
     public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler) {
-        return create(new OperationSkip.SkipTimed<T>(this, time, unit, scheduler));
+        return lift(new OperatorSkipTimed<T>(time, unit, scheduler));
     }
 
     /**

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

@@ -0,0 +1,80 @@
+/**
+ * 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 java.util.concurrent.atomic.AtomicBoolean;
+import rx.Observable.Operator;
+import rx.Scheduler;
+import rx.Scheduler.Worker;
+import rx.Subscriber;
+import rx.functions.Action0;
+
+/**
+ * Skips elements until a specified time elapses.
+ * @param <T> the value type
+ */
+public final class OperatorSkipTimed<T> implements Operator<T, T> {
+    final long time;
+    final TimeUnit unit;
+    final Scheduler scheduler;
+
+    public OperatorSkipTimed(long time, TimeUnit unit, Scheduler scheduler) {
+        this.time = time;
+        this.unit = unit;
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        final Worker worker = scheduler.createWorker();
+        child.add(worker);
+        final AtomicBoolean gate = new AtomicBoolean();
+        worker.schedule(new Action0() {
+            @Override
+            public void call() {
+                gate.set(true);
+            }
+        }, time, unit);
+        return new Subscriber<T>(child) {
+
+            @Override
+            public void onNext(T t) {
+                if (gate.get()) {
+                    child.onNext(t);
+                }
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                try {
+                    child.onError(e);
+                } finally {
+                    unsubscribe();
+                }
+            }
+
+            @Override
+            public void onCompleted() {
+                try {
+                    child.onCompleted();
+                } finally {
+                    unsubscribe();
+                }
+            }
+        };
+    }
+}

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

@@ -28,7 +28,7 @@
 
 import rx.Observable;
 import rx.Observer;
-import rx.operators.OperationSkipTest.CustomException;
+import rx.operators.OperatorSkipTimedTest.CustomException;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 

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

@@ -28,7 +28,7 @@
 
 import rx.Observable;
 import rx.Observer;
-import rx.operators.OperationSkipTest.CustomException;
+import rx.operators.OperatorSkipTimedTest.CustomException;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 

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

@@ -31,7 +31,7 @@
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 
-public class OperationSkipTest {
+public class OperatorSkipTimedTest {
 
     @Test
     public void testSkipTimed() {
@@ -41,6 +41,7 @@ public void testSkipTimed() {
 
         Observable<Integer> result = source.skip(1, TimeUnit.SECONDS, scheduler);
 
+        @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
         result.subscribe(o);
@@ -78,6 +79,7 @@ public void testSkipTimedFinishBeforeTime() {
 
         Observable<Integer> result = source.skip(1, TimeUnit.SECONDS, scheduler);
 
+        @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
         result.subscribe(o);
@@ -108,6 +110,7 @@ public void testSkipTimedErrorBeforeTime() {
 
         Observable<Integer> result = source.skip(1, TimeUnit.SECONDS, scheduler);
 
+        @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
         result.subscribe(o);
@@ -135,6 +138,7 @@ public void testSkipTimedErrorAfterTime() {
 
         Observable<Integer> result = source.skip(1, TimeUnit.SECONDS, scheduler);
 
+        @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
         result.subscribe(o);