1.x: Add to() conversion function to all stream types.

src/main/java/rx/Completable.java

@@ -2222,13 +2222,14 @@ public final Completable timeout0(long timeout, TimeUnit unit, Scheduler schedul
     }
 
     /**
-     * Allows fluent conversion to another type via a function callback.
-     * @param <U> the output type as determined by the converter function
-     * @param converter the function called with this which should return some other value.
-     * @return the converted value
-     * @throws NullPointerException if converter is null
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Single instance and returns a value
+     * @return the value returned by the function
      */
-    public final <U> U to(Func1<? super Completable, U> converter) {
+    public final <R> R to(Func1<? super Completable, R> converter) {
         return converter.call(this);
     }
 

src/main/java/rx/Observable.java

@@ -212,7 +212,7 @@ public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<
      * @return an instance of R created by the provided conversion function
      * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
      */
-    @Experimental
+    @Experimental @Deprecated // TODO remove method some time after 1.2.0 release. It was never a stable API.
     public <R> R extend(Func1<? super OnSubscribe<T>, ? extends R> conversion) {
         return conversion.call(new OnSubscribeExtend<T>(this));
     }
@@ -309,6 +309,19 @@ public interface Transformer<T, R> extends Func1<Observable<T>, Observable<R>> {
         // cover for generics insanity
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Observable instance and returns a value
+     * @return the value returned by the function
+     */
+    @Experimental
+    public final <R> R to(Func1<? super Observable<T>, R> converter) {
+        return converter.call(this);
+    }
+
     /**
      * Returns a Single that emits the single item emitted by the source Observable, if that Observable
      * emits only a single item. If the source Observable emits more than one item or no items, notify of an

src/main/java/rx/Single.java

@@ -2197,6 +2197,19 @@ public void onError(Throwable e) {
             }
         });
     }
+
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Single instance and returns a value
+     * @return the value returned by the function
+     */
+    @Experimental
+    public final <R> R to(Func1<? super Single<T>, R> converter) {
+        return converter.call(this);
+    }
 
     /**
      * Converts this Single into an {@link Observable}.

src/test/java/rx/CompletableTest.java

@@ -4158,4 +4158,21 @@ public Boolean call(Throwable t) {
         Assert.assertTrue(errors.get(1).toString(), errors.get(1) instanceof TestException);
         Assert.assertEquals(errors.get(1).toString(), "Forced inner failure", errors.get(1).getMessage());
     }
+
+    @Test public void toFunctionReceivesObservableReturnsResult() {
+        Completable c = Completable.error(new RuntimeException());
+
+        final Object expectedResult = new Object();
+        final AtomicReference<Completable> completableRef = new AtomicReference<Completable>();
+        Object actualResult = c.to(new Func1<Completable, Object>() {
+            @Override
+            public Object call(Completable completable) {
+                completableRef.set(completable);
+                return expectedResult;
+            }
+        });
+
+        assertSame(expectedResult, actualResult);
+        assertSame(c, completableRef.get());
+    }
 }

src/test/java/rx/ObservableTests.java

@@ -1423,4 +1423,21 @@ public void call() {
         assertTrue(list.get(1).toString(), list.get(1) instanceof TestException);
         assertEquals(100, list.get(2));
     }
+
+    @Test public void toFunctionReceivesObservableReturnsResult() {
+        Observable<String> o = Observable.just("Hi");
+
+        final Object expectedResult = new Object();
+        final AtomicReference<Observable<?>> observableRef = new AtomicReference<Observable<?>>();
+        Object actualResult = o.to(new Func1<Observable<String>, Object>() {
+            @Override
+            public Object call(Observable<String> observable) {
+                observableRef.set(observable);
+                return expectedResult;
+            }
+        });
+
+        assertSame(expectedResult, actualResult);
+        assertSame(o, observableRef.get());
+    }
 }

src/test/java/rx/SingleTest.java

@@ -2069,4 +2069,21 @@ public Completable call(final Integer integer) {
 
         testSubscriber.assertError(UnsupportedOperationException.class);
     }
+
+    @Test public void toFunctionReceivesObservableReturnsResult() {
+        Single<String> s = Single.just("Hi");
+
+        final Object expectedResult = new Object();
+        final AtomicReference<Single<?>> singleRef = new AtomicReference<Single<?>>();
+        Object actualResult = s.to(new Func1<Single<String>, Object>() {
+            @Override
+            public Object call(Single<String> single) {
+                singleRef.set(single);
+                return expectedResult;
+            }
+        });
+
+        assertSame(expectedResult, actualResult);
+        assertSame(s, singleRef.get());
+    }
 }