Change Lift to use rx.Observable.Operator

The generics insanity has to stop.

Author: benjchristensen

rxjava-contrib/rxjava-debug/src/main/java/rx/operators/DebugSubscriber.java

@@ -1,5 +1,6 @@
 package rx.operators;
 
+import rx.Observable.Operator;
 import rx.Observer;
 import rx.Subscriber;
 import rx.plugins.DebugNotification;

rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/DebugHook.java

@@ -2,10 +2,10 @@
 
 import rx.Observable;
 import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.Subscription;
 import rx.operators.DebugSubscriber;
-import rx.operators.Operator;
 import rx.util.functions.Action1;
 import rx.util.functions.Actions;
 import rx.util.functions.Func1;

rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/DebugNotification.java

@@ -2,11 +2,10 @@
 
 import rx.Notification;
 import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
 import rx.Observer;
 import rx.observers.SafeSubscriber;
 import rx.operators.DebugSubscriber;
-import rx.operators.Operator;
-import rx.plugins.DebugNotification.Kind;
 
 public class DebugNotification<T> {
     public static enum Kind {

rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/NotificationEvent.java

@@ -2,10 +2,10 @@
 
 import rx.Notification;
 import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
 import rx.Observer;
 import rx.observers.SafeSubscriber;
 import rx.operators.DebugSubscriber;
-import rx.operators.Operator;
 
 public class NotificationEvent<T> {
     public static enum Kind {

rxjava-contrib/rxjava-string/src/main/java/rx/observables/StringObservable.java

@@ -31,8 +31,8 @@
 
 import rx.Observable;
 import rx.Observable.OnSubscribe;
+import rx.Observable.Operator;
 import rx.Subscriber;
-import rx.operators.Operator;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 

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

@@ -94,7 +94,6 @@
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationUsing;
 import rx.operators.OperationWindow;
-import rx.operators.Operator;
 import rx.operators.OperatorCast;
 import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorFilter;
@@ -174,7 +173,7 @@ public class Observable<T> {
      *            {@link OnSubscribe} to be executed when {@link #subscribe(Subscriber)} is called
      */
     protected Observable(OnSubscribe<T> f) {
-        this.f = f;
+        this.f = hook.onCreate(f);
     }
 
     private final static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
@@ -206,19 +205,28 @@ protected Observable(OnSubscribe<T> f) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.create.aspx">MSDN: Observable.Create</a>
      */
     public final static <T> Observable<T> create(OnSubscribe<T> f) {
-        return new Observable<T>(hook.onCreate(f));
+        return new Observable<T>(f);
     }
 
     /**
-     * 
+     * Invoked when Obserable.subscribe is called. 
      */
     public static interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
-
+        // cover for generics insanity
     }
+    
+    /**
+     * Operator function for lifting into an Observable.
+     */
+    public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<? super T>> {
+        // cover for generics insanity
+    }
+
 
     /**
-     * 
+     * @Deprecated 
      */
+    @Deprecated
     public final static <T> Observable<T> create(final OnSubscribeFunc<T> f) {
         return new Observable<T>(new OnSubscribe<T>() {
 
@@ -253,11 +261,11 @@ public static interface OnSubscribeFunc<T> extends Function {
      * @param bind
      * @return an Observable that emits values that are the result of applying the bind function to the values of the current Observable
      */
-    public <R> Observable<R> lift(final Func1<Subscriber<? super R>, Subscriber<? super T>> bind) {
+    public <R> Observable<R> lift(final Operator<R, T> bind) {
         return new Observable<R>(new OnSubscribe<R>() {
             @Override
             public void call(Subscriber<? super R> o) {
-                subscribe(hook.onLift((Operator<R, T>) bind).call(o));
+                subscribe(hook.onLift(bind).call(o));
             }
         });
     }

rxjava-core/src/main/java/rx/joins/JoinObserver1.java

@@ -25,7 +25,6 @@
 import rx.Observable;
 import rx.Subscriber;
 import rx.observers.SafeSubscriber;
-import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Action1;
 
 /**

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -29,7 +29,6 @@
 import rx.operators.OperationNext;
 import rx.operators.OperationToFuture;
 import rx.operators.OperationToIterator;
-import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Action1;
 import rx.util.functions.Func1;
 

rxjava-core/src/main/java/rx/observers/Observers.java

@@ -1,7 +1,6 @@
 package rx.observers;
 
 import rx.Observer;
-import rx.Subscriber;
 import rx.util.OnErrorNotImplementedException;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;

rxjava-core/src/main/java/rx/observers/TestObserver.java

@@ -16,7 +16,6 @@
 package rx.observers;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;