3.x: Add Maybe.dematerialize (#6871)

akarnokd · web-flow · commit ea8ab3bbd561 · 2020-01-25T16:58:59.000+01:00
diff --git a/src/main/java/io/reactivex/rxjava3/core/Maybe.java b/src/main/java/io/reactivex/rxjava3/core/Maybe.java
@@ -2893,6 +2893,47 @@ public final Single<T> defaultIfEmpty(@NonNull T defaultItem) {
         return RxJavaPlugins.onAssembly(new MaybeToSingle<>(this, defaultItem));
     }
 
+    /**
+     * Maps the {@link Notification} success value of the current {@code Maybe} back into normal
+     * {@code onSuccess}, {@code onError} or {@code onComplete} signals.
+     * <p>
+     * <img width="640" height="268" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.dematerialize.png" alt="">
+     * <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.
+     * <p>
+     * Regular {@code onError} or {@code onComplete} signals from the current {@code Maybe} are passed along to the downstream.
+     * <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>
+     * Maybe.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 {@code Notification} instance.
+     * @return the new {@code Maybe} instance
+     * @throws NullPointerException if {@code selector} is {@code null}
+     * @since 3.0.0
+     * @see #materialize()
+     */
+    @CheckReturnValue
+    @NonNull
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <@NonNull R> Maybe<R> dematerialize(@NonNull Function<? super T, @NonNull Notification<R>> selector) {
+        Objects.requireNonNull(selector, "selector is null");
+        return RxJavaPlugins.onAssembly(new MaybeDematerialize<>(this, selector));
+    }
+
     /**
      * Returns a {@code Maybe} that signals the events emitted by the current {@code Maybe} shifted forward in time by a
      * specified delay.
diff --git a/src/main/java/io/reactivex/rxjava3/core/Single.java b/src/main/java/io/reactivex/rxjava3/core/Single.java
@@ -2511,7 +2511,7 @@ public final Single<T> delaySubscription(long time, @NonNull TimeUnit unit, @Non
     }
 
     /**
-     * Maps the {@link Notification} success value of this {@code Single} back into normal
+     * Maps the {@link Notification} success value of the current {@code Single} back into normal
      * {@code onSuccess}, {@code onError} or {@code onComplete} signals as a
      * {@link Maybe} source.
      * <p>
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeDematerialize.java b/src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeDematerialize.java
@@ -0,0 +1,108 @@
+/**
+ * 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
+ *
+ * https://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.rxjava3.internal.operators.maybe;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Function;
+import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
+
+import java.util.Objects;
+
+/**
+ * Maps the success value of the source to a Notification, then
+ * maps it back to the corresponding signal type.
+ * <p>History: 2.2.4 - experimental
+ * @param <T> the element type of the source
+ * @param <R> the element type of the Notification and result
+ * @since 3.0.0
+ */
+public final class MaybeDematerialize<T, R> extends AbstractMaybeWithUpstream<T, R> {
+
+    final Function<? super T, Notification<R>> selector;
+
+    public MaybeDematerialize(Maybe<T> source, Function<? super T, Notification<R>> selector) {
+        super(source);
+        this.selector = selector;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super R> observer) {
+        source.subscribe(new DematerializeObserver<>(observer, selector));
+    }
+
+    static final class DematerializeObserver<T, R> implements MaybeObserver<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 = Objects.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);
+        }
+
+        @Override
+        public void onComplete() {
+            downstream.onComplete();
+        }
+    }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeDematerializeTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeDematerializeTest.java
@@ -0,0 +1,122 @@
+/**
+ * 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
+ *
+ * https://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.rxjava3.internal.operators.maybe;
+
+import static org.mockito.Mockito.*;
+import org.junit.Test;
+
+import io.reactivex.rxjava3.core.*;
+import io.reactivex.rxjava3.exceptions.TestException;
+import io.reactivex.rxjava3.functions.Function;
+import io.reactivex.rxjava3.internal.functions.Functions;
+import io.reactivex.rxjava3.subjects.MaybeSubject;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class MaybeDematerializeTest extends RxJavaTest {
+
+    @Test
+    public void success() {
+        Maybe.just(Notification.createOnNext(1))
+        .dematerialize(Functions.<Notification<Integer>>identity())
+        .test()
+        .assertResult(1);
+    }
+
+    @Test
+    public void empty() {
+        Maybe.just(Notification.<Integer>createOnComplete())
+        .dematerialize(Functions.<Notification<Integer>>identity())
+        .test()
+        .assertResult();
+    }
+
+    @Test
+    public void emptySource() throws Throwable {
+        @SuppressWarnings("unchecked")
+        Function<Notification<Integer>, Notification<Integer>> function = mock(Function.class);
+
+        Maybe.<Notification<Integer>>empty()
+        .dematerialize(function)
+        .test()
+        .assertResult();
+
+        verify(function, never()).apply(any());
+    }
+
+    @Test
+    public void error() {
+        Maybe.<Notification<Integer>>error(new TestException())
+        .dematerialize(Functions.<Notification<Integer>>identity())
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void errorNotification() {
+        Maybe.just(Notification.<Integer>createOnError(new TestException()))
+        .dematerialize(Functions.<Notification<Integer>>identity())
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void doubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeMaybe(new Function<Maybe<Object>, MaybeSource<Object>>() {
+            @SuppressWarnings({ "unchecked", "rawtypes" })
+            @Override
+            public MaybeSource<Object> apply(Maybe<Object> v) throws Exception {
+                return v.dematerialize((Function)Functions.identity());
+            }
+        });
+    }
+
+    @Test
+    public void dispose() {
+        TestHelper.checkDisposed(MaybeSubject.<Notification<Integer>>create().dematerialize(Functions.<Notification<Integer>>identity()));
+    }
+
+    @Test
+    public void selectorCrash() {
+        Maybe.just(Notification.createOnNext(1))
+        .dematerialize(new Function<Notification<Integer>, Notification<Integer>>() {
+            @Override
+            public Notification<Integer> apply(Notification<Integer> v) throws Exception {
+                throw new TestException();
+            }
+        })
+        .test()
+        .assertFailure(TestException.class);
+    }
+
+    @Test
+    public void selectorNull() {
+        Maybe.just(Notification.createOnNext(1))
+        .dematerialize(Functions.justFunction((Notification<Integer>)null))
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+
+    @Test
+    public void selectorDifferentType() {
+        Maybe.just(Notification.createOnNext(1))
+        .dematerialize(new Function<Notification<Integer>, Notification<String>>() {
+            @Override
+            public Notification<String> apply(Notification<Integer> v) throws Exception {
+                return Notification.createOnNext("Value-" + 1);
+            }
+        })
+        .test()
+        .assertResult("Value-1");
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -2511,7 +2511,7 @@ public final Single<T> delaySubscription(long time, @NonNull TimeUnit unit, @Non`
`2511`	`2511`	`}`
`2512`	`2512`
`2513`	`2513`	`/**`
`2514`		`- * Maps the {@link Notification} success value of this {@code Single} back into normal`
	`2514`	`+ * Maps the {@link Notification} success value of the current {@code Single} back into normal`
`2515`	`2515`	`* {@code onSuccess}, {@code onError} or {@code onComplete} signals as a`
`2516`	`2516`	`* {@link Maybe} source.`
`2517`	`2517`	`* <p>`