Add OnSubscribeFunction and refactor to support it

This is related to ongoing work related to covariant support at #331

Author: benjchristensen

language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/GroovyOnSubscribeFuncWrapper.java

@@ -0,0 +1,41 @@
+/**
+ * Copyright 2013 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.lang.groovy;
+
+import groovy.lang.Closure;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+
+/**
+ * Concrete wrapper that accepts a {@link Closure} and produces a {@link OnSubscribeFunc}.
+ * 
+ * @param <T>
+ */
+public class GroovyOnSubscribeFuncWrapper<T> implements OnSubscribeFunc<T> {
+
+    private final Closure<Subscription> closure;
+
+    public GroovyOnSubscribeFuncWrapper(Closure<Subscription> closure) {
+        this.closure = closure;
+    }
+
+    @Override
+    public Subscription call(Observer<? super T> observer) {
+        return closure.call(observer);
+    }
+
+}

language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/RxGroovyExtensionModule.java

@@ -33,6 +33,7 @@
 import org.codehaus.groovy.runtime.metaclass.MetaClassRegistryImpl;
 
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
 import rx.observables.BlockingObservable;
 import rx.util.functions.Action;
 import rx.util.functions.Function;
@@ -52,33 +53,6 @@ public RxGroovyExtensionModule() {
         super("RxGroovyExtensionModule", "1.0");
     }
 
-    /**
-     * Keeping this code around a little while as it was hard to figure out ... and I'm still messing with it while debugging.
-     * 
-     * Once the rest of this ExtensionModule stuff is working I'll delete this method.
-     * 
-     * This is used for manually initializing rather than going via the org.codehaus.groovy.runtime.ExtensionModule properties file.
-     */
-    public static void initializeManuallyForTesting() {
-        System.out.println("initialize");
-        MetaClassRegistryImpl mcRegistry = ((MetaClassRegistryImpl) GroovySystem.getMetaClassRegistry());
-        //        RxGroovyExtensionModule em = new RxGroovyExtensionModule();
-
-        Properties p = new Properties();
-        p.setProperty("moduleFactory", "rx.lang.groovy.RxGroovyPropertiesModuleFactory");
-        Map<CachedClass, List<MetaMethod>> metaMethods = new HashMap<CachedClass, List<MetaMethod>>();
-        mcRegistry.registerExtensionModuleFromProperties(p, RxGroovyExtensionModule.class.getClassLoader(), metaMethods);
-
-        for (ExtensionModule m : mcRegistry.getModuleRegistry().getModules()) {
-            System.out.println("Module: " + m.getName());
-        }
-
-        for (CachedClass cc : metaMethods.keySet()) {
-            System.out.println("Adding MetaMethods to CachedClass: " + cc);
-            cc.addNewMopMethods(metaMethods.get(cc));
-        }
-    }
-
     @SuppressWarnings("rawtypes")
     @Override
     public List<MetaMethod> getMetaMethods() {
@@ -135,6 +109,8 @@ public Object invoke(Object object, Object[] arguments) {
                         if (o instanceof Closure) {
                             if (Action.class.isAssignableFrom(m.getParameterTypes()[i])) {
                                 newArgs[i] = new GroovyActionWrapper((Closure) o);
+                            } else if(OnSubscribeFunc.class.isAssignableFrom(m.getParameterTypes()[i])) {
+                                newArgs[i] = new GroovyOnSubscribeFuncWrapper((Closure) o);
                             } else {
                                 newArgs[i] = new GroovyFunctionWrapper((Closure) o);
                             }

language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy

@@ -31,6 +31,7 @@ import org.mockito.MockitoAnnotations;
 
 import rx.Notification;
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
 import rx.observables.GroupedObservable;
@@ -296,7 +297,7 @@ def class ObservableTests {
     }
 
 
-    def class AsyncObservable implements Func1<Observer<Integer>, Subscription> {
+    def class AsyncObservable implements OnSubscribeFunc {
 
         public Subscription call(final Observer<Integer> observer) {
             new Thread(new Runnable() {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/AbstractButtonSource.java

@@ -15,10 +15,7 @@
  */
 package rx.swing.sources;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
 import java.awt.event.ActionEvent;
 import java.awt.event.ActionListener;
@@ -29,20 +26,21 @@
 import org.mockito.Matchers;
 
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
+import rx.observables.SwingObservable;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
-import rx.util.functions.Func1;
 
 public enum AbstractButtonSource { ; // no instances
 
     /**
      * @see SwingObservable.fromButtonAction
      */
     public static Observable<ActionEvent> fromActionOf(final AbstractButton button) {
-        return Observable.create(new Func1<Observer<? super ActionEvent>, Subscription>() {
+        return Observable.create(new OnSubscribeFunc<ActionEvent>() {
             @Override
             public Subscription call(final Observer<? super ActionEvent> observer) {
                 final ActionListener listener = new ActionListener() {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/ComponentEventSource.java

@@ -15,14 +15,15 @@
  */
 package rx.swing.sources;
 
-import static rx.swing.sources.ComponentEventSource.Predicate.RESIZED;
+import static rx.swing.sources.ComponentEventSource.Predicate.*;
 
 import java.awt.Component;
 import java.awt.Dimension;
 import java.awt.event.ComponentEvent;
 import java.awt.event.ComponentListener;
 
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
 import rx.observables.SwingObservable;
@@ -36,7 +37,7 @@ public enum ComponentEventSource { ; // no instances
      * @see SwingObservable.fromComponentEvents
      */
     public static Observable<ComponentEvent> fromComponentEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<? super ComponentEvent>, Subscription>() {
+        return Observable.create(new OnSubscribeFunc<ComponentEvent>() {
             @Override
             public Subscription call(final Observer<? super ComponentEvent> observer) {
                 final ComponentListener listener = new ComponentListener() {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/KeyEventSource.java

@@ -15,12 +15,8 @@
  */
 package rx.swing.sources;
 
-import static java.util.Arrays.asList;
-import static org.mockito.Mockito.inOrder;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static java.util.Arrays.*;
+import static org.mockito.Mockito.*;
 
 import java.awt.Component;
 import java.awt.event.KeyEvent;
@@ -36,12 +32,12 @@
 import org.mockito.Matchers;
 
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
-import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
 public enum KeyEventSource { ; // no instances
@@ -50,7 +46,7 @@ public enum KeyEventSource { ; // no instances
      * @see SwingObservable.fromKeyEvents(Component)
      */
     public static Observable<KeyEvent> fromKeyEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<? super KeyEvent>, Subscription>() {
+        return Observable.create(new OnSubscribeFunc<KeyEvent>() {
             @Override
             public Subscription call(final Observer<? super KeyEvent> observer) {
                 final KeyListener listener = new KeyListener() {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/MouseEventSource.java

@@ -21,19 +21,20 @@
 import java.awt.event.MouseMotionListener;
 
 import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
+import rx.observables.SwingObservable;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
-import rx.util.functions.Func1;
 
 public enum MouseEventSource { ; // no instances
 
     /**
      * @see SwingObservable.fromMouseEvents
      */
     public static Observable<MouseEvent> fromMouseEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<? super MouseEvent>, Subscription>() {
+        return Observable.create(new OnSubscribeFunc<MouseEvent>() {
             @Override
             public Subscription call(final Observer<? super MouseEvent> observer) {
                 final MouseListener listener = new MouseListener() {
@@ -78,7 +79,7 @@ public void call() {
      * @see SwingObservable.fromMouseMotionEvents
      */
     public static Observable<MouseEvent> fromMouseMotionEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<? super MouseEvent>, Subscription>() {
+        return Observable.create(new OnSubscribeFunc<MouseEvent>() {
             @Override
             public Subscription call(final Observer<? super MouseEvent> observer) {
                 final MouseMotionListener listener = new MouseMotionListener() {

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

@@ -113,8 +113,19 @@ public class Observable<T> {
 
     private final static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
 
-    private final Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe;
+    private final OnSubscribeFunc<T> onSubscribe;
 
+    /**
+     * Function interface for work to be performed when an {@link Observable} is subscribed to via {@link Observable#subscribe(Observer)}
+     * 
+     * @param <T>
+     */
+    public static interface OnSubscribeFunc<T> extends Function<T> {
+
+        public Subscription call(Observer<? super T> t1);
+
+    }
+    
     /**
      * Observable with Function to execute when subscribed to.
      * <p>
@@ -124,7 +135,7 @@ public class Observable<T> {
      * @param onSubscribe
      *            {@link Func1} to be executed when {@link #subscribe(Observer)} is called.
      */
-    protected Observable(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
+    protected Observable(OnSubscribeFunc<T> onSubscribe) {
         this.onSubscribe = onSubscribe;
     }
 
@@ -162,7 +173,7 @@ protected Observable() {
      */
     public Subscription subscribe(Observer<? super T> observer) {
         // allow the hook to intercept and/or decorate
-        Func1<? super Observer<? super T>, ? extends Subscription> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
+        OnSubscribeFunc<T> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
         // validate and proceed
         if (observer == null) {
             throw new IllegalArgumentException("observer can not be null");
@@ -402,7 +413,7 @@ private void handleError(Throwable e) {
      */
     private static class NeverObservable<T> extends Observable<T> {
         public NeverObservable() {
-            super(new Func1<Observer<? super T>, Subscription>() {
+            super(new OnSubscribeFunc<T>() {
 
                 @Override
                 public Subscription call(Observer<? super T> t1) {
@@ -422,7 +433,7 @@ public Subscription call(Observer<? super T> t1) {
     private static class ThrowObservable<T> extends Observable<T> {
 
         public ThrowObservable(final Throwable exception) {
-            super(new Func1<Observer<? super T>, Subscription>() {
+            super(new OnSubscribeFunc<T>() {
 
                 /**
                  * Accepts an {@link Observer} and calls its {@link Observer#onError onError} method.
@@ -466,7 +477,7 @@ public Subscription call(Observer<? super T> observer) {
      * @return an Observable that, when an {@link Observer} subscribes to it, will execute the given
      *         function
      */
-    public static <T> Observable<T> create(Func1<? super Observer<? super T>, ? extends Subscription> func) {
+    public static <T> Observable<T> create(OnSubscribeFunc<T> func) {
         return new Observable<T>(func);
     }
 

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

@@ -60,7 +60,7 @@
  */
 public class BlockingObservable<T> extends Observable<T> {
 
-    protected BlockingObservable(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
+    protected BlockingObservable(OnSubscribeFunc<T> onSubscribe) {
         super(onSubscribe);
     }
 
@@ -76,7 +76,7 @@ private BlockingObservable() {
      * Convert an Observable into a BlockingObservable.
      */
     public static <T> BlockingObservable<T> from(final Observable<? extends T> o) {
-        return new BlockingObservable<T>(new Func1<Observer<? super T>, Subscription>() {
+        return new BlockingObservable<T>(new OnSubscribeFunc<T>() {
 
             @Override
             public Subscription call(Observer<? super T> observer) {
@@ -784,7 +784,7 @@ public void testToIterable() {
 
         @Test(expected = TestException.class)
         public void testToIterableWithException() {
-            BlockingObservable<String> obs = BlockingObservable.from(create(new Func1<Observer<? super String>, Subscription>() {
+            BlockingObservable<String> obs = BlockingObservable.from(create(new OnSubscribeFunc<String>() {
 
                 @Override
                 public Subscription call(Observer<? super String> observer) {
@@ -807,7 +807,7 @@ public Subscription call(Observer<? super String> observer) {
         @Test
         public void testForEachWithError() {
             try {
-                BlockingObservable.from(Observable.create(new Func1<Observer<? super String>, Subscription>() {
+                BlockingObservable.from(Observable.create(new OnSubscribeFunc<String>() {
 
                     @Override
                     public Subscription call(final Observer<? super String> observer) {

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

@@ -18,7 +18,6 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
-import rx.util.functions.Func1;
 
 /**
  * A ConnectableObservable resembles an ordinary {@link Observable}, except that it does not begin
@@ -37,7 +36,7 @@
 
 public abstract class ConnectableObservable<T> extends Observable<T> {
 
-    protected ConnectableObservable(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
+    protected ConnectableObservable(OnSubscribeFunc<T> onSubscribe) {
         super(onSubscribe);
     }