convert to OnSubscribe

Author: davidmoten

src/main/java/rx/Observable.java

@@ -9203,7 +9203,7 @@ public final Observable<T> skip(long time, TimeUnit unit) {
      * @see <a href="http://reactivex.io/documentation/operators/skip.html">ReactiveX operators documentation: Skip</a>
      */
     public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler) {
-        return lift(new OperatorSkipTimed<T>(time, unit, scheduler));
+        return create(new OnSubscribeSkipTimed<T>(this, time, unit, scheduler));
     }
 
     /**

src/main/java/rx/internal/operators/OperatorSkipTimed.java renamed to src/main/java/rx/internal/operators/OnSubscribeSkipTimed.java

@@ -17,7 +17,8 @@
 
 import java.util.concurrent.TimeUnit;
 
-import rx.Observable.Operator;
+import rx.Observable.OnSubscribe;
+import rx.Observable;
 import rx.Scheduler;
 import rx.Scheduler.Worker;
 import rx.Subscriber;
@@ -27,24 +28,26 @@
  * Skips elements until a specified time elapses.
  * @param <T> the value type
  */
-public final class OperatorSkipTimed<T> implements Operator<T, T> {
+public final class OnSubscribeSkipTimed<T> implements OnSubscribe<T> {
     final long time;
     final TimeUnit unit;
     final Scheduler scheduler;
+    final Observable<T> source;
 
-    public OperatorSkipTimed(long time, TimeUnit unit, Scheduler scheduler) {
+    public OnSubscribeSkipTimed(Observable<T> source, long time, TimeUnit unit, Scheduler scheduler) {
+        this.source = source;
         this.time = time;
         this.unit = unit;
         this.scheduler = scheduler;
     }
 
     @Override
-    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+    public void call(final Subscriber<? super T> child) {
         final Worker worker = scheduler.createWorker();
         child.add(worker);
         SkipTimedSubscriber<T> subscriber = new SkipTimedSubscriber<T>(child);
         worker.schedule(subscriber, time, unit);
-        return subscriber;
+        source.unsafeSubscribe(subscriber);
     }
 
     final static class SkipTimedSubscriber<T> extends Subscriber<T> implements Action0 {