Merge pull request #770 from benjchristensen/bind

Bind Operator

Author: benjchristensen

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

@@ -0,0 +1,43 @@
+/**
+ * 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.Operator;
+import rx.Subscription;
+import rx.util.functions.Action1;
+
+public class GroovyCreateWrapper<T> implements Action1<Operator<? super T>> {
+
+    private final Closure<Void> closure;
+
+    public GroovyCreateWrapper(Closure<Void> closure) {
+        this.closure = closure;
+    }
+
+    @Override
+    public void call(Operator<? super T> op) {
+        Object o = closure.call(op);
+        /*
+         * If the new signature is being used, we will get NULL back.
+         * If the old is being used we will get a Subscription back.
+         */
+        if (o != null) {
+            op.add((Subscription) o);
+        }
+    }
+
+}

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

@@ -16,21 +16,16 @@
 package rx.lang.groovy;
 
 import groovy.lang.Closure;
-import groovy.lang.GroovySystem;
 import groovy.lang.MetaMethod;
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Properties;
 
 import org.codehaus.groovy.reflection.CachedClass;
 import org.codehaus.groovy.reflection.ReflectionCache;
 import org.codehaus.groovy.runtime.m12n.ExtensionModule;
-import org.codehaus.groovy.runtime.metaclass.MetaClassRegistryImpl;
 
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
@@ -75,6 +70,9 @@ public List<MetaMethod> getMetaMethods() {
     }
 
     private MetaMethod createMetaMethod(final Method m) {
+        if (m.getDeclaringClass().equals(Observable.class) && m.getName().equals("create")) {
+            return specialCasedOverrideForCreate(m);
+        }
         return new MetaMethod() {
 
             @Override
@@ -109,12 +107,11 @@ 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])) {
+                            } else if (OnSubscribeFunc.class.isAssignableFrom(m.getParameterTypes()[i])) {
                                 newArgs[i] = new GroovyOnSubscribeFuncWrapper((Closure) o);
                             } else {
                                 newArgs[i] = new GroovyFunctionWrapper((Closure) o);
                             }
-
                         } else {
                             newArgs[i] = o;
                         }
@@ -152,4 +149,55 @@ public CachedClass[] getParameterTypes() {
             }
         };
     }
+
+    /**
+     * Special case until we finish migrating off the deprecated 'create' method signature
+     */
+    private MetaMethod specialCasedOverrideForCreate(final Method m) {
+        return new MetaMethod() {
+
+            @Override
+            public int getModifiers() {
+                return m.getModifiers();
+            }
+
+            @Override
+            public String getName() {
+                return m.getName();
+            }
+
+            @Override
+            public Class getReturnType() {
+                return m.getReturnType();
+            }
+
+            @Override
+            public CachedClass getDeclaringClass() {
+                return ReflectionCache.getCachedClass(m.getDeclaringClass());
+            }
+
+            @Override
+            public Object invoke(Object object, final Object[] arguments) {
+                return Observable.create(new GroovyCreateWrapper((Closure) arguments[0]));
+            }
+
+            @SuppressWarnings("rawtypes")
+            @Override
+            public CachedClass[] getParameterTypes() {
+                Class[] pts = m.getParameterTypes();
+                CachedClass[] cc = new CachedClass[pts.length];
+                for (int i = 0; i < pts.length; i++) {
+                    if (Function.class.isAssignableFrom(pts[i])) {
+                        // function type to be replaced by closure
+                        cc[i] = ReflectionCache.getCachedClass(Closure.class);
+                    } else {
+                        // non-function type
+                        cc[i] = ReflectionCache.getCachedClass(pts[i]);
+                    }
+                }
+                return cc;
+            }
+        };
+    }
+
 }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -310,9 +310,8 @@ trait Observable[+T]
    *         their index. Indices start at 0.
    */
   def zipWithIndex: Observable[(T, Int)] = {
-    val fScala: (T, Integer) => (T, Int) = (elem: T, index: Integer) => (elem, index)
-    val fJava : Func2[_ >: T, Integer, _ <: (T, Int)] = fScala
-    toScalaObservable[(T, Int)](asJavaObservable.mapWithIndex[(T, Int)](fJava))
+    var n = 0;
+    this.map(x => { val result = (x,n); n += 1; result })
   }
 
   /**