3.x: Add Single.ofType (#6876)

Author: akarnokd

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

@@ -3472,7 +3472,7 @@ public final Maybe<T> doOnSuccess(@NonNull Consumer<? super T> onSuccess) {
      * Filters the success item of the {@code Maybe} via a predicate function and emitting it if the predicate
      * returns {@code true}, completing otherwise.
      * <p>
-     * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/filter.png" alt="">
+     * <img width="640" height="291" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.filter.png" alt="">
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code filter} does not operate by default on a particular {@link Scheduler}.</dd>
@@ -4098,10 +4098,10 @@ public final Maybe<T> observeOn(@NonNull Scheduler scheduler) {
     }
 
     /**
-     * Filters the items emitted by a {@code Maybe}, only emitting its success value if that
+     * Filters the items emitted by the current {@code Maybe}, only emitting its success value if that
      * is an instance of the supplied {@link Class}.
      * <p>
-     * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/ofClass.png" alt="">
+     * <img width="640" height="291" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.ofType.png" alt="">
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code ofType} does not operate by default on a particular {@link Scheduler}.</dd>

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

@@ -3439,6 +3439,30 @@ public final Single<Boolean> contains(@NonNull Object item, @NonNull BiPredicate
     public final Flowable<T> mergeWith(@NonNull SingleSource<? extends T> other) {
         return merge(this, other);
     }
+    /**
+     * Filters the items emitted by the current {@code Single}, only emitting its success value if that
+     * is an instance of the supplied {@link Class}.
+     * <p>
+     * <img width="640" height="399" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.ofType.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code ofType} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <U> the output type
+     * @param clazz
+     *            the class type to filter the items emitted by the current {@code Single}
+     * @return the new {@link Maybe} instance
+     * @throws NullPointerException if {@code clazz} is {@code null}
+     * @see <a href="http://reactivex.io/documentation/operators/filter.html">ReactiveX operators documentation: Filter</a>
+     */
+    @CheckReturnValue
+    @NonNull
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <U> Maybe<U> ofType(@NonNull Class<U> clazz) {
+        Objects.requireNonNull(clazz, "clazz is null");
+        return filter(Functions.isInstanceOf(clazz)).cast(clazz);
+    }
 
     /**
      * Signals the success item or the terminal signals of the current {@code Single} on the specified {@link Scheduler},

src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleOfTypeTest.java

@@ -0,0 +1,96 @@
+/**
+ * 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.rxjava3.internal.operators.single;
+
+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.observers.TestObserver;
+import io.reactivex.rxjava3.processors.PublishProcessor;
+import io.reactivex.rxjava3.testsupport.TestHelper;
+
+public class SingleOfTypeTest extends RxJavaTest {
+
+    @Test
+    public void normal() {
+        Single.just(1).ofType(Integer.class)
+        .test()
+        .assertResult(1);
+    }
+
+    @Test
+    public void normalDowncast() {
+        TestObserver<Number> to = Single.just(1)
+        .ofType(Number.class)
+        .test();
+        // don't make this fluent, target type required!
+        to.assertResult((Number)1);
+    }
+
+    @Test
+    public void notInstance() {
+        TestObserver<String> to = Single.just(1)
+        .ofType(String.class)
+        .test();
+        // don't make this fluent, target type required!
+        to.assertResult();
+    }
+
+    @Test
+    public void error() {
+        TestObserver<Number> to = Single.<Integer>error(new TestException())
+        .ofType(Number.class)
+        .test();
+        // don't make this fluent, target type required!
+        to.assertFailure(TestException.class);
+    }
+
+    @Test
+    public void errorNotInstance() {
+        TestObserver<String> to = Single.<Integer>error(new TestException())
+        .ofType(String.class)
+        .test();
+        // don't make this fluent, target type required!
+        to.assertFailure(TestException.class);
+    }
+
+    @Test
+    public void dispose() {
+        TestHelper.checkDisposedSingleToMaybe(new Function<Single<Object>, Maybe<Object>>() {
+            @Override
+            public Maybe<Object> apply(Single<Object> m) throws Exception {
+                return m.ofType(Object.class);
+            }
+        });
+    }
+
+    @Test
+    public void isDisposed() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+
+        TestHelper.checkDisposed(pp.singleElement().ofType(Object.class));
+    }
+
+    @Test
+    public void doubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeSingleToMaybe(new Function<Single<Object>, Maybe<Object>>() {
+            @Override
+            public Maybe<Object> apply(Single<Object> f) throws Exception {
+                return f.ofType(Object.class);
+            }
+        });
+    }
+}

src/test/java/io/reactivex/rxjava3/testsupport/TestHelper.java

@@ -2248,6 +2248,30 @@ public static <T, U> void checkDisposedMaybeToSingle(Function<Maybe<T>, ? extend
         assertFalse(pp.hasSubscribers());
     }
 
+    /**
+     * Check if the operator applied to a Maybe source propagates dispose properly.
+     * @param <T> the source value type
+     * @param <U> the output value type
+     * @param composer the function to apply an operator to the provided Maybe source
+     */
+    public static <T, U> void checkDisposedSingleToMaybe(Function<Single<T>, ? extends MaybeSource<U>> composer) {
+        PublishProcessor<T> pp = PublishProcessor.create();
+
+        TestSubscriber<U> ts = new TestSubscriber<>();
+
+        try {
+            new MaybeToFlowable<>(composer.apply(pp.singleOrError())).subscribe(ts);
+        } catch (Throwable ex) {
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+
+        assertTrue(pp.hasSubscribers());
+
+        ts.cancel();
+
+        assertFalse(pp.hasSubscribers());
+    }
+
     /**
      * Check if the TestSubscriber has a CompositeException with the specified class
      * of Throwables in the given order.