Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

3.x: Add Single.ofType #6876

Merged
merged 1 commit into from
Jan 26, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions src/main/java/io/reactivex/rxjava3/core/Maybe.java
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand Down
24 changes: 24 additions & 0 deletions src/main/java/io/reactivex/rxjava3/core/Single.java
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand Down
Original file line number Diff line number Diff line change
@@ -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);
}
});
}
}
24 changes: 24 additions & 0 deletions src/test/java/io/reactivex/rxjava3/testsupport/TestHelper.java
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down