Add Object overloads for dynamic language support.

Need these until we finish work at ReactiveX#204

Author: benjchristensen

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

@@ -19,17 +19,21 @@ import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
 
 import org.junit.Before;
 import org.junit.Test;
 import static org.junit.Assert.*;
+
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
 import rx.Notification;
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.observables.GroupedObservable;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func1;
 
@@ -280,6 +284,29 @@ def class ObservableTests {
         Observable.from(1, 2, 3).all({ x -> x > 0 }).subscribe({ result -> a.received(result) });
         verify(a, times(1)).received(true);
     }
+    
+    @Test
+    public void testGroupBy() {
+        int count=0;
+        
+        Observable.from("one", "two", "three", "four", "five", "six")
+        .groupBy({String s -> s.length()})
+        .mapMany({
+            groupObservable ->
+            
+            return groupObservable.map({
+                s ->
+                return "Value: " + s + " Group: " + groupObservable.getKey(); 
+            });
+          }).toBlockingObservable().forEach({
+            s ->
+            println(s);
+            count++;
+          })
+          
+          assertEquals(6, count);
+    }
+    
 
     def class AsyncObservable implements Func1<Observer<Integer>, Subscription> {
 

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

@@ -1294,6 +1294,28 @@ public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T
         return create(OperationGroupBy.groupBy(source, keySelector, elementSelector));
     }
 
+    @SuppressWarnings("rawtypes")
+    public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T> source, final Object keySelector, final Object elementSelector) {
+        final FuncN _k = Functions.from(keySelector);
+        final FuncN _e = Functions.from(elementSelector);
+
+        return groupBy(source, new Func1<T, K>() {
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public K call(T t1) {
+                return (K) _k.call(t1);
+            }
+        }, new Func1<T, R>() {
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public R call(T t1) {
+                return (R) _e.call(t1);
+            }
+        });
+    }
+
     /**
      * Groups the items emitted by an Observable according to a specified criteria, and emits these
      * grouped items as Observables, one Observable per group.
@@ -1314,6 +1336,20 @@ public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T
     public static <K, T> Observable<GroupedObservable<K, T>> groupBy(Observable<T> source, final Func1<T, K> keySelector) {
         return create(OperationGroupBy.groupBy(source, keySelector));
     }
+    
+    @SuppressWarnings("rawtypes")
+    public static <K, T> Observable<GroupedObservable<K, T>> groupBy(Observable<T> source, final Object keySelector) {
+        final FuncN _k = Functions.from(keySelector);
+
+        return groupBy(source, new Func1<T, K>() {
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public K call(T t1) {
+                return (K) _k.call(t1);
+            }
+        });
+    }
 
     /**
      * This behaves like <code>merge</code> except that if any of the merged Observables notify
@@ -3560,6 +3596,10 @@ public Observable<T> startWith(T... values) {
     public <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<T, K> keySelector, final Func1<T, R> elementSelector) {
         return groupBy(this, keySelector, elementSelector);
     }
+    
+    public <K, R> Observable<GroupedObservable<K, R>> groupBy(final Object keySelector, final Object elementSelector) {
+        return groupBy(this, keySelector, elementSelector);
+    }
 
     /**
      * Groups the items emitted by an Observable according to a specified criteria, and emits these
@@ -3578,6 +3618,10 @@ public <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<T, K> keyS
     public <K> Observable<GroupedObservable<K, T>> groupBy(final Func1<T, K> keySelector) {
         return groupBy(this, keySelector);
     }
+    
+    public <K> Observable<GroupedObservable<K, T>> groupBy(final Object keySelector) {
+        return groupBy(this, keySelector);
+    }
 
     /**
      * Converts an Observable into a BlockingObservable (an Observable with blocking operators).