Compose/Transformer

See #1416 for discussion that led to this.

This does not have co/contra-variance which needs to be figured out.

Author: benjchristensen

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

@@ -179,9 +179,34 @@ public void call(Subscriber<? super R> o) {
             }
         });
     }
+    
+    
+    /**
+     * Compose Observables together with a function.
+     * 
+     * This works on the Observables themselves whereas `lift` works on the internal Subscriber/Observers.
+     * 
+     * Lift should be used when creating an operator that acts on the underlying data. 
+     * Compose should be used when acting on the observable itself, such as composing multiple operators.
+     * 
+     * @param transformer
+     * @return
+     */
+    public <R> Observable<R> compose(Transformer<T, R> transformer) {
+        return transformer.call(this);
+    }
+    
+    /**
+     * Transformer function for `compose`
+     */
+    public static interface Transformer<T, R> extends Func1<Observable<T>, Observable<R>> {
+        // cover for generics insanity
+    }
+    
+    
 
     /* *********************************************************************************************************
-     * Observers Below Here
+     * Operators Below Here
      * *********************************************************************************************************
      */
 

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -46,6 +46,7 @@
 import org.mockito.MockitoAnnotations;
 
 import rx.Observable.OnSubscribe;
+import rx.Observable.Transformer;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action1;
 import rx.functions.Action2;
@@ -1100,4 +1101,27 @@ public void call(List<Boolean> booleans) {
         }
         assertEquals(nums.length, count.get());
     }
+    
+    @Test
+    public void testCompose() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable.from(1, 2, 3).compose(new Transformer<Integer, String>() {
+
+            @Override
+            public Observable<String> call(Observable<Integer> t1) {
+                return t1.map(new Func1<Integer, String>() {
+
+                    @Override
+                    public String call(Integer t1) {
+                        return String.valueOf(t1);
+                    }
+                    
+                });
+            }
+            
+        }).subscribe(ts);
+        ts.assertTerminalEvent();
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Arrays.asList("1", "2", "3"));
+    }
 }