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

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

* Add remaining test cases

* Correct dematerialize javadoc

* Use dematerialize selector fix some docs

Author: akarnokd

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,27 @@ 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
+     * @see Single#dematerialize(Function)
+     */
+    @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,26 @@ 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
+     * @see Single#dematerialize(Function)
+     */
+    @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,43 @@ 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>
+     * The intended use of the {@code selector} function is to perform a
+     * type-safe identity mapping (see example) on a source that is already of type
+     * {@code Notification<T>}. The Java language doesn't allow
+     * limiting instance methods to a certain generic argument shape, therefore,
+     * a function is used to ensure the conversion remains type safe.
+     * <dl>
+     * <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code dematerialize} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * <p>
+     * Example:
+     * <pre><code>
+     * Single.just(Notification.createOnNext(1))
+     * .dematerialize(notification -&gt; notification)
+     * .test()
+     * .assertResult(1);
+     * </code></pre>
+     * @param <R> the result type
+     * @param selector the function called with the success item and should
+     * return a {@link Notification} instance.
+     * @return the new Maybe instance
+     * @since 2.2.4 - experimental
+     * @see #materialize()
+     */
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final <R> Maybe<R> dematerialize(Function<? super T, Notification<R>> selector) {
+        ObjectHelper.requireNonNull(selector, "selector is null");
+        return RxJavaPlugins.onAssembly(new SingleDematerialize<T, R>(this, selector));
+    }
+
     /**
      * Calls the specified consumer with the success item after this item has been emitted to the downstream.
      * <p>
@@ -2871,6 +2908,26 @@ 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
+     * @see #dematerialize(Function)
+     */
+    @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,105 @@
+/**
+ * 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.exceptions.Exceptions;
+import io.reactivex.functions.Function;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.functions.ObjectHelper;
+
+/**
+ * Maps the success value of the source to a Notification, then
+ * maps it back to the corresponding signal type.
+ * @param <T> the element type of the source
+ * @param <R> the element type of the Notification and result
+ * @since 2.2.4 - experimental
+ */
+@Experimental
+public final class SingleDematerialize<T, R> extends Maybe<R> {
+
+    final Single<T> source;
+
+    final Function<? super T, Notification<R>> selector;
+
+    public SingleDematerialize(Single<T> source, Function<? super T, Notification<R>> selector) {
+        this.source = source;
+        this.selector = selector;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super R> observer) {
+        source.subscribe(new DematerializeObserver<T, R>(observer, selector));
+    }
+
+    static final class DematerializeObserver<T, R> implements SingleObserver<T>, Disposable {
+
+        final MaybeObserver<? super R> downstream;
+
+        final Function<? super T, Notification<R>> selector;
+
+        Disposable upstream;
+
+        DematerializeObserver(MaybeObserver<? super R> downstream,
+                Function<? super T, Notification<R>> selector) {
+            this.downstream = downstream;
+            this.selector = selector;
+        }
+
+        @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(T t) {
+            Notification<R> notification;
+
+            try {
+                notification = ObjectHelper.requireNonNull(selector.apply(t), "The selector returned a null Notification");
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                downstream.onError(ex);
+                return;
+            }
+            if (notification.isOnNext()) {
+                downstream.onSuccess(notification.getValue());
+            } else if (notification.isOnComplete()) {
+                downstream.onComplete();
+            } else {
+                downstream.onError(notification.getError());
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            downstream.onError(e);
+        }
+    }
+}