2.x: Implement as()

src/main/java/io/reactivex/Completable.java

@@ -908,6 +908,32 @@ public final Completable andThen(CompletableSource next) {
         return concatWith(next);
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code as} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Completable instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if converter is null
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> R as(@NonNull CompletableConverter<? extends R> converter) {
+        try {
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+    }
+
     /**
      * Subscribes to and awaits the termination of this Completable instance in a blocking manner and
      * rethrows any exception emitted.

src/main/java/io/reactivex/CompletableConverter.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex;
+
+import io.reactivex.annotations.*;
+
+/**
+ * Convenience interface and callback used by the {@link Completable#as} operator to turn a Completable into another
+ * value fluently.
+ *
+ * @param <R> the output type
+ */
+@Experimental
+public interface CompletableConverter<R> {
+    /**
+     * Applies a function to the upstream Completable and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream Completable instance
+     * @return the converted value
+     * @throws Exception on error
+     */
+    @NonNull
+    R apply(@NonNull Completable upstream) throws Exception;
+}

src/main/java/io/reactivex/Flowable.java

@@ -5237,6 +5237,35 @@ public final Single<Boolean> any(Predicate<? super T> predicate) {
         return RxJavaPlugins.onAssembly(new FlowableAnySingle<T>(this, predicate));
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The backpressure behavior depends on what happens in the {@code converter} function.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code as} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Flowable instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if converter is null
+     */
+    @Experimental
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.SPECIAL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> R as(@NonNull FlowableConverter<T, ? extends R> converter) {
+        try {
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+    }
+
     /**
      * Returns the first item emitted by this {@code Flowable}, or throws
      * {@code NoSuchElementException} if it emits no items.

src/main/java/io/reactivex/FlowableConverter.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex;
+
+import io.reactivex.annotations.*;
+
+/**
+ * Convenience interface and callback used by the {@link Flowable#as} operator to turn a Flowable into another
+ * value fluently.
+ *
+ * @param <T> the upstream type
+ * @param <R> the output type
+ */
+@Experimental
+public interface FlowableConverter<T, R> {
+    /**
+     * Applies a function to the upstream Flowable and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream Flowable instance
+     * @return the converted value
+     * @throws Exception on error
+     */
+    @NonNull
+    R apply(@NonNull Flowable<T> upstream) throws Exception;
+}

src/main/java/io/reactivex/Maybe.java

@@ -1989,6 +1989,32 @@ public final Maybe<T> ambWith(MaybeSource<? extends T> other) {
         return ambArray(this, other);
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code as} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Maybe instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if converter is null
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> R as(@NonNull MaybeConverter<T, ? extends R> converter) {
+        try {
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+    }
+
     /**
      * Waits in a blocking fashion until the current Maybe signals a success value (which is returned),
      * null if completed or an exception (which is propagated).

src/main/java/io/reactivex/MaybeConverter.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex;
+
+import io.reactivex.annotations.*;
+
+/**
+ * Convenience interface and callback used by the {@link Maybe#as} operator to turn a Maybe into another
+ * value fluently.
+ *
+ * @param <T> the upstream type
+ * @param <R> the output type
+ */
+@Experimental
+public interface MaybeConverter<T, R> {
+    /**
+     * Applies a function to the upstream Maybe and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream Maybe instance
+     * @return the converted value
+     * @throws Exception on error
+     */
+    @NonNull
+    R apply(@NonNull Maybe<T> upstream) throws Exception;
+}

src/main/java/io/reactivex/Observable.java

@@ -4800,6 +4800,32 @@ public final Single<Boolean> any(Predicate<? super T> predicate) {
         return RxJavaPlugins.onAssembly(new ObservableAnySingle<T>(this, predicate));
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code as} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Observable instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if converter is null
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> R as(@NonNull ObservableConverter<T, ? extends R> converter) {
+        try {
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+    }
+
     /**
      * Returns the first item emitted by this {@code Observable}, or throws
      * {@code NoSuchElementException} if it emits no items.

src/main/java/io/reactivex/ObservableConverter.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex;
+
+import io.reactivex.annotations.*;
+
+/**
+ * Convenience interface and callback used by the {@link Observable#as} operator to turn an Observable into another
+ * value fluently.
+ *
+ * @param <T> the upstream type
+ * @param <R> the output type
+ */
+@Experimental
+public interface ObservableConverter<T, R> {
+    /**
+     * Applies a function to the upstream Observable and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream Observable instance
+     * @return the converted value
+     * @throws Exception on error
+     */
+    @NonNull
+    R apply(@NonNull Observable<T> upstream) throws Exception;
+}

src/main/java/io/reactivex/Single.java

@@ -1522,6 +1522,32 @@ public final Single<T> ambWith(SingleSource<? extends T> other) {
         return ambArray(this, other);
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code as} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current Single instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if converter is null
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <R> R as(@NonNull SingleConverter<T, ? extends R> converter) {
+        try {
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+    }
+
     /**
      * Hides the identity of the current Single, including the Disposable that is sent
      * to the downstream via {@code onSubscribe()}.

src/main/java/io/reactivex/SingleConverter.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex;
+
+import io.reactivex.annotations.*;
+
+/**
+ * Convenience interface and callback used by the {@link Single#as} operator to turn a Single into another
+ * value fluently.
+ *
+ * @param <T> the upstream type
+ * @param <R> the output type
+ */
+@Experimental
+public interface SingleConverter<T, R> {
+    /**
+     * Applies a function to the upstream Single and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream Single instance
+     * @return the converted value
+     * @throws Exception on error
+     */
+    @NonNull
+    R apply(@NonNull Single<T> upstream) throws Exception;
+}

src/test/java/io/reactivex/completable/CompletableTest.java

@@ -2796,11 +2796,43 @@ public void accept(Object e) { }
         });
     }
 
+    @Test(timeout = 5000)
+    public void asNormal() {
+        Flowable<Object> flow = normal.completable.as(new CompletableConverter<Flowable<Object>>() {
+            @Override
+            public Flowable<Object> apply(Completable c) {
+                return c.toFlowable();
+            }
+        });
+
+        flow.blockingForEach(new Consumer<Object>() {
+            @Override
+            public void accept(Object e) { }
+        });
+    }
+
+    @Test
+    public void as() {
+        Completable.complete().as(new CompletableConverter<Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(Completable v) throws Exception {
+                return v.toFlowable();
+            }
+        })
+        .test()
+        .assertComplete();
+    }
+
     @Test(expected = NullPointerException.class)
     public void toNull() {
         normal.completable.to(null);
     }
 
+    @Test(expected = NullPointerException.class)
+    public void asNull() {
+        normal.completable.as(null);
+    }
+
     @Test(timeout = 5000)
     public void toFlowableNormal() {
         normal.completable.toFlowable().blockingForEach(Functions.emptyConsumer());