2.x: Add materialize() and dematerialize()

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

@@ -26,7 +26,7 @@
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.operators.maybe.*;
 import io.reactivex.internal.operators.mixed.*;
-import io.reactivex.internal.operators.single.SingleDelayWithCompletable;
+import io.reactivex.internal.operators.single.*;
 import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -1782,6 +1782,26 @@ public final Completable lift(final CompletableOperator onLift) {
         return RxJavaPlugins.onAssembly(new CompletableLift(this, onLift));
     }
 
+    /**
+     * Maps the signal types of this Completable into a {@link Notification} of the same kind
+     * and emits it as a single success value to downstream.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/materialize.png" alt="">
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code materialize} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param <T> the intended target element type of the notification
+     * @return the new Single instance
+     * @since 2.2.4 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <T> Single<Notification<T>> materialize() {
+        return RxJavaPlugins.onAssembly(new CompletableMaterialize<T>(this));
+    }
+
     /**
      * Returns a Completable which subscribes to this and the other Completable and completes
      * when both of them complete or one emits an error.

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

@@ -3377,6 +3377,25 @@ public final <R> Maybe<R> map(Function<? super T, ? extends R> mapper) {
         return RxJavaPlugins.onAssembly(new MaybeMap<T, R>(this, mapper));
     }
 
+    /**
+     * Maps the signal types of this Maybe into a {@link Notification} of the same kind
+     * and emits it as a single success value to downstream.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/materialize.png" alt="">
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code materialize} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @return the new Single instance
+     * @since 2.2.4 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Single<Notification<T>> materialize() {
+        return RxJavaPlugins.onAssembly(new MaybeMaterialize<T>(this));
+    }
+
     /**
      * Flattens this and another Maybe into a single Flowable, without any transformation.
      * <p>

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

@@ -2302,6 +2302,44 @@ public final Single<T> delaySubscription(long time, TimeUnit unit, Scheduler sch
         return delaySubscription(Observable.timer(time, unit, scheduler));
     }
 
+    /**
+     * Maps the {@link Notification} success value of this Single back into normal
+     * {@code onSuccess}, {@code onError} or {@code onComplete} signals as a
+     * {@link Maybe} source.
+     * <p>
+     * Note that {@code this} should be of type {@code Single<Notification<T>>} or
+     * the transformation will result in an {@code onError} signal of
+     * {@link ClassCastException}. Currently, the Java language doesn't allow specifying
+     * methods for certain type argument shapes only (unlike extension methods would),
+     * hence the forced casting in this operator.
+     * <p>
+     * In addition, usually the inner value type (T2) has to be expressed again via
+     * a type argument on this method (see example below).
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code dematerialize} does by default subscribe to the current Single
+     * on the {@link Scheduler} you provided, after the delay.</dd>
+     * </dl>
+     * <p>
+     * Example:
+     * <pre><code>
+     * Single.just(Notification.createOnNext(1))
+     * .&lt;Integer&gt;dematerialize()
+     * .test()
+     * .assertResult(1);
+     * </code></pre>
+     * @param <T2> the type inside the Notification
+     * @return the new Maybe instance
+     * @since 2.2.4 - experimental
+     */
+    @SuppressWarnings("unchecked")
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final <T2> Maybe<T2> dematerialize() {
+        return RxJavaPlugins.onAssembly(new SingleDematerialize<T2>((Single<Object>)this));
+    }
+
     /**
      * Calls the specified consumer with the success item after this item has been emitted to the downstream.
      * <p>
@@ -2871,6 +2909,25 @@ public final <R> Single<R> map(Function<? super T, ? extends R> mapper) {
         return RxJavaPlugins.onAssembly(new SingleMap<T, R>(this, mapper));
     }
 
+    /**
+     * Maps the signal types of this Single into a {@link Notification} of the same kind
+     * and emits it as a single success value to downstream.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/materialize.png" alt="">
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     * <dd>{@code materialize} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @return the new Single instance
+     * @since 2.2.4 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Single<Notification<T>> materialize() {
+        return RxJavaPlugins.onAssembly(new SingleMaterialize<T>(this));
+    }
+
     /**
      * Signals true if the current Single signals a success value that is Object-equals with the value
      * provided.

src/main/java/io/reactivex/internal/operators/completable/CompletableMaterialize.java

@@ -0,0 +1,40 @@
+/**
+ * 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.internal.operators.completable;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.internal.operators.mixed.MaterializeSingleObserver;
+
+/**
+ * Turn the signal types of a Completable source into a single Notification of
+ * equal kind.
+ *
+ * @param <T> the element type of the source
+ * @since 2.2.4 - experimental
+ */
+@Experimental
+public final class CompletableMaterialize<T> extends Single<Notification<T>> {
+
+    final Completable source;
+
+    public CompletableMaterialize(Completable source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super Notification<T>> observer) {
+        source.subscribe(new MaterializeSingleObserver<T>(observer));
+    }
+}

src/main/java/io/reactivex/internal/operators/maybe/MaybeMaterialize.java

@@ -0,0 +1,40 @@
+/**
+ * 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.internal.operators.maybe;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.internal.operators.mixed.MaterializeSingleObserver;
+
+/**
+ * Turn the signal types of a Maybe source into a single Notification of
+ * equal kind.
+ *
+ * @param <T> the element type of the source
+ * @since 2.2.4 - experimental
+ */
+@Experimental
+public final class MaybeMaterialize<T> extends Single<Notification<T>> {
+
+    final Maybe<T> source;
+
+    public MaybeMaterialize(Maybe<T> source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super Notification<T>> observer) {
+        source.subscribe(new MaterializeSingleObserver<T>(observer));
+    }
+}

src/main/java/io/reactivex/internal/operators/mixed/MaterializeSingleObserver.java

@@ -0,0 +1,71 @@
+/**
+ * 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.internal.operators.mixed;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+/**
+ * A consumer that implements the consumer types of Maybe, Single and Completable
+ * and turns their signals into Notifications for a SingleObserver.
+ * @param <T> the element type of the source
+ * @since 2.2.4 - experimental
+ */
+@Experimental
+public final class MaterializeSingleObserver<T>
+implements SingleObserver<T>, MaybeObserver<T>, CompletableObserver, Disposable {
+
+    final SingleObserver<? super Notification<T>> downstream;
+
+    Disposable upstream;
+
+    public MaterializeSingleObserver(SingleObserver<? super Notification<T>> downstream) {
+        this.downstream = downstream;
+    }
+
+    @Override
+    public void onSubscribe(Disposable d) {
+        if (DisposableHelper.validate(upstream, d)) {
+            this.upstream = d;
+            downstream.onSubscribe(this);
+        }
+    }
+
+    @Override
+    public void onComplete() {
+        downstream.onSuccess(Notification.<T>createOnComplete());
+    }
+
+    @Override
+    public void onSuccess(T t) {
+        downstream.onSuccess(Notification.<T>createOnNext(t));
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        downstream.onSuccess(Notification.<T>createOnError(e));
+    }
+
+    @Override
+    public boolean isDisposed() {
+        return upstream.isDisposed();
+    }
+
+    @Override
+    public void dispose() {
+        upstream.dispose();
+    }
+}

src/main/java/io/reactivex/internal/operators/single/SingleDematerialize.java

@@ -0,0 +1,91 @@
+/**
+ * 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.internal.operators.single;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+/**
+ * Maps the Notification success value of the source back to normal
+ * onXXX method call as a Maybe.
+ * @param <T> the element type of the notification and result
+ * @since 2.2.4 - experimental
+ */
+@Experimental
+public final class SingleDematerialize<T> extends Maybe<T> {
+
+    final Single<Object> source;
+
+    public SingleDematerialize(Single<Object> source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super T> observer) {
+        source.subscribe(new DematerializeObserver<T>(observer));
+    }
+
+    static final class DematerializeObserver<T> implements SingleObserver<Object>, Disposable {
+
+        final MaybeObserver<? super T> downstream;
+
+        Disposable upstream;
+
+        DematerializeObserver(MaybeObserver<? super T> downstream) {
+            this.downstream = downstream;
+        }
+
+        @Override
+        public void dispose() {
+            upstream.dispose();
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return upstream.isDisposed();
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(upstream, d)) {
+                upstream = d;
+                downstream.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(Object t) {
+            if (t instanceof Notification) {
+                @SuppressWarnings("unchecked")
+                Notification<T> notification = (Notification<T>)t;
+                if (notification.isOnNext()) {
+                    downstream.onSuccess(notification.getValue());
+                } else if (notification.isOnComplete()) {
+                    downstream.onComplete();
+                } else {
+                    downstream.onError(notification.getError());
+                }
+            } else {
+                downstream.onError(new ClassCastException("io.reactivex.Notification expected but got " + t.getClass()));
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            downstream.onError(e);
+        }
+    }
+}