diff --git a/src/main/java/io/reactivex/rxjava3/core/Completable.java b/src/main/java/io/reactivex/rxjava3/core/Completable.java index 05b3f8b240..96eb10c447 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Completable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Completable.java @@ -19,7 +19,7 @@ import org.reactivestreams.*; import io.reactivex.rxjava3.annotations.*; -import io.reactivex.rxjava3.disposables.Disposable; +import io.reactivex.rxjava3.disposables.*; import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.*; @@ -29,7 +29,7 @@ import io.reactivex.rxjava3.internal.operators.completable.*; import io.reactivex.rxjava3.internal.operators.maybe.*; import io.reactivex.rxjava3.internal.operators.mixed.*; -import io.reactivex.rxjava3.internal.operators.single.*; +import io.reactivex.rxjava3.internal.operators.single.SingleDelayWithCompletable; import io.reactivex.rxjava3.observers.TestObserver; import io.reactivex.rxjava3.plugins.RxJavaPlugins; import io.reactivex.rxjava3.schedulers.Schedulers; @@ -2836,6 +2836,7 @@ public final Completable hide() { *
{@code subscribe} does not operate by default on a particular {@link Scheduler}.
* * @return the new {@code Disposable} that can be used for disposing the subscription at any time + * @see #subscribe(Action, Consumer, DisposableContainer) */ @SchedulerSupport(SchedulerSupport.NONE) @NonNull @@ -2921,6 +2922,7 @@ public final void subscribe(@NonNull CompletableObserver observer) { * @param onError the {@link Consumer} that is called if this {@code Completable} emits an error * @return the new {@link Disposable} that can be used for disposing the subscription at any time * @throws NullPointerException if {@code onComplete} or {@code onError} is {@code null} + * @see #subscribe(Action, Consumer, DisposableContainer) */ @CheckReturnValue @NonNull @@ -2934,6 +2936,44 @@ public final Disposable subscribe(@NonNull Action onComplete, @NonNull Consumer< return observer; } + /** + * Wraps the given onXXX callbacks into a {@link Disposable} {@link CompletableObserver}, + * adds it to the given {@link DisposableContainer} and ensures, that if the upstream + * terminates or this particular {@code Disposable} is disposed, the {@code CompletableObserver} is removed + * from the given composite. + *

+ * The {@code CompletableObserver} will be removed after the callback for the terminal event has been invoked. + *

+ *
Scheduler:
+ *
{@code subscribe} does not operate by default on a particular {@link Scheduler}.
+ *
+ * @param onError the callback for an upstream error + * @param onComplete the callback for an upstream completion + * @param container the {@code DisposableContainer} (such as {@link CompositeDisposable}) to add and remove the + * created {@code Disposable} {@code CompletableObserver} + * @return the {@code Disposable} that allows disposing the particular subscription. + * @throws NullPointerException + * if {@code onComplete}, {@code onError} + * or {@code container} is {@code null} + * @since 3.1.0 + */ + @SchedulerSupport(SchedulerSupport.NONE) + @NonNull + public final Disposable subscribe( + @NonNull Action onComplete, + @NonNull Consumer onError, + @NonNull DisposableContainer container) { + Objects.requireNonNull(onComplete, "onComplete is null"); + Objects.requireNonNull(onError, "onError is null"); + Objects.requireNonNull(container, "container is null"); + + DisposableAutoReleaseMultiObserver observer = new DisposableAutoReleaseMultiObserver<>( + container, Functions.emptyConsumer(), onError, onComplete); + container.add(observer); + subscribe(observer); + return observer; + } + /** * Subscribes to this {@code Completable} and calls the given {@link Action} when this {@code Completable} * completes normally. @@ -2950,6 +2990,7 @@ public final Disposable subscribe(@NonNull Action onComplete, @NonNull Consumer< * @param onComplete the {@code Action} called when this {@code Completable} completes normally * @return the new {@link Disposable} that can be used for disposing the subscription at any time * @throws NullPointerException if {@code onComplete} is {@code null} + * @see #subscribe(Action, Consumer, DisposableContainer) */ @CheckReturnValue @NonNull diff --git a/src/main/java/io/reactivex/rxjava3/core/Flowable.java b/src/main/java/io/reactivex/rxjava3/core/Flowable.java index 804a057533..d22d8dc4dc 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Flowable.java @@ -20,7 +20,7 @@ import org.reactivestreams.*; import io.reactivex.rxjava3.annotations.*; -import io.reactivex.rxjava3.disposables.Disposable; +import io.reactivex.rxjava3.disposables.*; import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.flowables.*; import io.reactivex.rxjava3.functions.*; @@ -15699,6 +15699,7 @@ public final Flowable startWithArray(@NonNull T... items) { * * @return the new {@link Disposable} instance that allows cancelling the flow * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @BackpressureSupport(BackpressureKind.UNBOUNDED_IN) @SchedulerSupport(SchedulerSupport.NONE) @@ -15727,6 +15728,7 @@ public final Disposable subscribe() { * @throws NullPointerException * if {@code onNext} is {@code null} * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @BackpressureSupport(BackpressureKind.UNBOUNDED_IN) @@ -15753,9 +15755,10 @@ public final Disposable subscribe(@NonNull Consumer onNext) { * the {@code Consumer} you have designed to accept any error notification from the * current {@code Flowable} * @return the new {@link Disposable} instance that allows cancelling the flow - * @see ReactiveX operators documentation: Subscribe * @throws NullPointerException * if {@code onNext} or {@code onError} is {@code null} + * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @BackpressureSupport(BackpressureKind.UNBOUNDED_IN) @@ -15788,6 +15791,7 @@ public final Disposable subscribe(@NonNull Consumer onNext, @NonNull * @throws NullPointerException * if {@code onNext}, {@code onError} or {@code onComplete} is {@code null} * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @BackpressureSupport(BackpressureKind.UNBOUNDED_IN) @@ -15806,6 +15810,51 @@ public final Disposable subscribe(@NonNull Consumer onNext, @NonNull return ls; } + /** + * Wraps the given onXXX callbacks into a {@link Disposable} {@link Subscriber}, + * adds it to the given {@link DisposableContainer} and ensures, that if the upstream + * terminates or this particular {@code Disposable} is disposed, the {@code Subscriber} is removed + * from the given container. + *

+ * The {@coded Subscriber} will be removed after the callback for the terminal event has been invoked. + *

+ *
Backpressure:
+ *
The operator consumes the current {@code Flowable} in an unbounded manner (i.e., no + * backpressure is applied to it).
+ *
Scheduler:
+ *
{@code subscribe} does not operate by default on a particular {@link Scheduler}.
+ *
+ * @param onNext the callback for upstream items + * @param onError the callback for an upstream error if any + * @param onComplete the callback for the upstream completion if any + * @param container the {@code DisposableContainer} (such as {@link CompositeDisposable}) to add and remove the + * created {@code Disposable} {@code Subscriber} + * @return the {@code Disposable} that allows disposing the particular subscription. + * @throws NullPointerException + * if {@code onNext}, {@code onError}, + * {@code onComplete} or {@code container} is {@code null} + * @since 3.1.0 + */ + @BackpressureSupport(BackpressureKind.SPECIAL) + @SchedulerSupport(SchedulerSupport.NONE) + @NonNull + public final Disposable subscribe( + @NonNull Consumer onNext, + @NonNull Consumer onError, + @NonNull Action onComplete, + @NonNull DisposableContainer container) { + Objects.requireNonNull(onNext, "onNext is null"); + Objects.requireNonNull(onError, "onError is null"); + Objects.requireNonNull(onComplete, "onComplete is null"); + Objects.requireNonNull(container, "container is null"); + + DisposableAutoReleaseSubscriber subscriber = new DisposableAutoReleaseSubscriber<>( + container, onNext, onError, onComplete); + container.add(subscriber); + subscribe(subscriber); + return subscriber; + } + @BackpressureSupport(BackpressureKind.SPECIAL) @SchedulerSupport(SchedulerSupport.NONE) @Override diff --git a/src/main/java/io/reactivex/rxjava3/core/Maybe.java b/src/main/java/io/reactivex/rxjava3/core/Maybe.java index c0b8421fb1..917a9838a4 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Maybe.java +++ b/src/main/java/io/reactivex/rxjava3/core/Maybe.java @@ -20,7 +20,7 @@ import org.reactivestreams.*; import io.reactivex.rxjava3.annotations.*; -import io.reactivex.rxjava3.disposables.Disposable; +import io.reactivex.rxjava3.disposables.*; import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.*; @@ -5226,6 +5226,7 @@ public final Flowable startWith(@NonNull Publisher other) { * * @return the new {@link Disposable} instance that can be used for disposing the subscription at any time * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @SchedulerSupport(SchedulerSupport.NONE) @NonNull @@ -5250,6 +5251,7 @@ public final Disposable subscribe() { * @throws NullPointerException * if {@code onSuccess} is {@code null} * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @@ -5272,10 +5274,11 @@ public final Disposable subscribe(@NonNull Consumer onSuccess) { * the {@code Consumer} you have designed to accept any error notification from the * {@code Maybe} * @return the new {@link Disposable} instance that can be used for disposing the subscription at any time - * @see ReactiveX operators documentation: Subscribe * @throws NullPointerException * if {@code onSuccess} is {@code null}, or * if {@code onError} is {@code null} + * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @@ -5305,6 +5308,7 @@ public final Disposable subscribe(@NonNull Consumer onSuccess, @NonNu * if {@code onSuccess}, {@code onError} or * {@code onComplete} is {@code null} * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @NonNull @@ -5317,6 +5321,47 @@ public final Disposable subscribe(@NonNull Consumer onSuccess, @NonNu return subscribeWith(new MaybeCallbackObserver<>(onSuccess, onError, onComplete)); } + /** + * Wraps the given onXXX callbacks into a {@link Disposable} {@link MaybeObserver}, + * adds it to the given {@link DisposableContainer} and ensures, that if the upstream + * terminates or this particular {@code Disposable} is disposed, the {@code MaybeObserver} is removed + * from the given composite. + *

+ * The {@code MaybeObserver} will be removed after the callback for the terminal event has been invoked. + *

+ *
Scheduler:
+ *
{@code subscribe} does not operate by default on a particular {@link Scheduler}.
+ *
+ * @param onSuccess the callback for upstream items + * @param onError the callback for an upstream error + * @param onComplete the callback for an upstream completion without any value or error + * @param container the {@code DisposableContainer} (such as {@link CompositeDisposable}) to add and remove the + * created {@code Disposable} {@code MaybeObserver} + * @return the {@code Disposable} that allows disposing the particular subscription. + * @throws NullPointerException + * if {@code onSuccess}, {@code onError}, + * {@code onComplete} or {@code container} is {@code null} + * @since 3.1.0 + */ + @SchedulerSupport(SchedulerSupport.NONE) + @NonNull + public final Disposable subscribe( + @NonNull Consumer onSuccess, + @NonNull Consumer onError, + @NonNull Action onComplete, + @NonNull DisposableContainer container) { + Objects.requireNonNull(onSuccess, "onSuccess is null"); + Objects.requireNonNull(onError, "onError is null"); + Objects.requireNonNull(onComplete, "onComplete is null"); + Objects.requireNonNull(container, "container is null"); + + DisposableAutoReleaseMultiObserver observer = new DisposableAutoReleaseMultiObserver<>( + container, onSuccess, onError, onComplete); + container.add(observer); + subscribe(observer); + return observer; + } + @SchedulerSupport(SchedulerSupport.NONE) @Override public final void subscribe(@NonNull MaybeObserver observer) { diff --git a/src/main/java/io/reactivex/rxjava3/core/Observable.java b/src/main/java/io/reactivex/rxjava3/core/Observable.java index 27d46ad8c5..18534eecc7 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Observable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Observable.java @@ -20,7 +20,7 @@ import org.reactivestreams.Publisher; import io.reactivex.rxjava3.annotations.*; -import io.reactivex.rxjava3.disposables.Disposable; +import io.reactivex.rxjava3.disposables.*; import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.*; @@ -13025,6 +13025,7 @@ public final Observable startWithArray(@NonNull T... items) { * * @return the new {@link Disposable} instance that can be used to dispose the subscription at any time * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @SchedulerSupport(SchedulerSupport.NONE) @NonNull @@ -13049,6 +13050,7 @@ public final Disposable subscribe() { * @throws NullPointerException * if {@code onNext} is {@code null} * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @@ -13071,9 +13073,10 @@ public final Disposable subscribe(@NonNull Consumer onNext) { * the {@code Consumer} you have designed to accept any error notification from the current * {@code Observable} * @return the new {@link Disposable} instance that can be used to dispose the subscription at any time - * @see ReactiveX operators documentation: Subscribe * @throws NullPointerException * if {@code onNext} or {@code onError} is {@code null} + * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @@ -13102,6 +13105,7 @@ public final Disposable subscribe(@NonNull Consumer onNext, @NonNull * @throws NullPointerException * if {@code onNext}, {@code onError} or {@code onComplete} is {@code null} * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, Action, DisposableContainer) */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @@ -13119,6 +13123,47 @@ public final Disposable subscribe(@NonNull Consumer onNext, @NonNull return ls; } + /** + * Wraps the given onXXX callbacks into a {@link Disposable} {@link Observer}, + * adds it to the given {@code DisposableContainer} and ensures, that if the upstream + * terminates or this particular {@code Disposable} is disposed, the {@code Observer} is removed + * from the given container. + *

+ * The {@code Observer} will be removed after the callback for the terminal event has been invoked. + *

+ *
Scheduler:
+ *
{@code subscribe} does not operate by default on a particular {@link Scheduler}.
+ *
+ * @param onNext the callback for upstream items + * @param onError the callback for an upstream error if any + * @param onComplete the callback for the upstream completion if any + * @param container the {@code DisposableContainer} (such as {@link CompositeDisposable}) to add and remove the + * created {@code Disposable} {@code Observer} + * @return the {@code Disposable} that allows disposing the particular subscription. + * @throws NullPointerException + * if {@code onNext}, {@code onError}, + * {@code onComplete} or {@code container} is {@code null} + * @since 3.1.0 + */ + @SchedulerSupport(SchedulerSupport.NONE) + @NonNull + public final Disposable subscribe( + @NonNull Consumer onNext, + @NonNull Consumer onError, + @NonNull Action onComplete, + @NonNull DisposableContainer container) { + Objects.requireNonNull(onNext, "onNext is null"); + Objects.requireNonNull(onError, "onError is null"); + Objects.requireNonNull(onComplete, "onComplete is null"); + Objects.requireNonNull(container, "container is null"); + + DisposableAutoReleaseObserver observer = new DisposableAutoReleaseObserver<>( + container, onNext, onError, onComplete); + container.add(observer); + subscribe(observer); + return observer; + } + @SchedulerSupport(SchedulerSupport.NONE) @Override public final void subscribe(@NonNull Observer observer) { diff --git a/src/main/java/io/reactivex/rxjava3/core/Single.java b/src/main/java/io/reactivex/rxjava3/core/Single.java index dbe3694913..66e1ffd3d5 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Single.java +++ b/src/main/java/io/reactivex/rxjava3/core/Single.java @@ -20,7 +20,7 @@ import org.reactivestreams.*; import io.reactivex.rxjava3.annotations.*; -import io.reactivex.rxjava3.disposables.Disposable; +import io.reactivex.rxjava3.disposables.*; import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.*; @@ -4705,6 +4705,7 @@ public final Flowable startWith(@NonNull Publisher other) { * * @return the new {@link Disposable} instance that can be used for disposing the subscription at any time * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, DisposableContainer) */ @SchedulerSupport(SchedulerSupport.NONE) @NonNull @@ -4726,9 +4727,10 @@ public final Disposable subscribe() { * the callback that receives either the success value or the failure {@link Throwable} * (whichever is not {@code null}) * @return the new {@link Disposable} instance that can be used for disposing the subscription at any time - * @see ReactiveX operators documentation: Subscribe * @throws NullPointerException * if {@code onCallback} is {@code null} + * @see #subscribe(Consumer, Consumer, DisposableContainer) + * @see ReactiveX operators documentation: Subscribe */ @CheckReturnValue @NonNull @@ -4760,6 +4762,7 @@ public final Disposable subscribe(@NonNull BiConsumerReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, DisposableContainer) */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @@ -4784,9 +4787,10 @@ public final Disposable subscribe(@NonNull Consumer onSuccess) { * the {@code Consumer} you have designed to accept any error notification from the * {@code Single} * @return the new {@link Disposable} instance that can be used for disposing the subscription at any time - * @see ReactiveX operators documentation: Subscribe * @throws NullPointerException * if {@code onSuccess} or {@code onError} is {@code null} + * @see ReactiveX operators documentation: Subscribe + * @see #subscribe(Consumer, Consumer, DisposableContainer) */ @CheckReturnValue @NonNull @@ -4800,6 +4804,44 @@ public final Disposable subscribe(@NonNull Consumer onSuccess, @NonNu return observer; } + /** + * Wraps the given onXXX callbacks into a {@link Disposable} {@link SingleObserver}, + * adds it to the given {@link DisposableContainer} and ensures, that if the upstream + * terminates or this particular {@code Disposable} is disposed, the {@code SingleObserver} is removed + * from the given container. + *

+ * The {@code SingleObserver} will be removed after the callback for the terminal event has been invoked. + *

+ *
Scheduler:
+ *
{@code subscribe} does not operate by default on a particular {@link Scheduler}.
+ *
+ * @param onSuccess the callback for upstream items + * @param onError the callback for an upstream error if any + * @param container the {@code DisposableContainer} (such as {@link CompositeDisposable}) to add and remove the + * created {@code Disposable} {@code SingleObserver} + * @return the {@code Disposable} that allows disposing the particular subscription. + * @throws NullPointerException + * if {@code onSuccess}, {@code onError} + * or {@code container} is {@code null} + * @since 3.1.0 + */ + @SchedulerSupport(SchedulerSupport.NONE) + @NonNull + public final Disposable subscribe( + @NonNull Consumer onSuccess, + @NonNull Consumer onError, + @NonNull DisposableContainer container) { + Objects.requireNonNull(onSuccess, "onSuccess is null"); + Objects.requireNonNull(onError, "onError is null"); + Objects.requireNonNull(container, "container is null"); + + DisposableAutoReleaseMultiObserver observer = new DisposableAutoReleaseMultiObserver<>( + container, onSuccess, onError, Functions.EMPTY_ACTION); + container.add(observer); + subscribe(observer); + return observer; + } + @SchedulerSupport(SchedulerSupport.NONE) @Override public final void subscribe(@NonNull SingleObserver observer) { diff --git a/src/main/java/io/reactivex/rxjava3/internal/observers/AbstractDisposableAutoRelease.java b/src/main/java/io/reactivex/rxjava3/internal/observers/AbstractDisposableAutoRelease.java new file mode 100644 index 0000000000..e5d89a4502 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava3/internal/observers/AbstractDisposableAutoRelease.java @@ -0,0 +1,125 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import java.util.concurrent.atomic.AtomicReference; + +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.*; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.disposables.DisposableHelper; +import io.reactivex.rxjava3.internal.functions.Functions; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; + +/** + * Wraps lambda callbacks and when the upstream terminates or the observer gets disposed, + * removes itself from a {@link io.reactivex.rxjava3.disposables.CompositeDisposable}. + *

History: 0.18.0 @ RxJavaExtensions + * @since 3.1.0 + */ +abstract class AbstractDisposableAutoRelease +extends AtomicReference +implements Disposable, LambdaConsumerIntrospection { + + private static final long serialVersionUID = 8924480688481408726L; + + final AtomicReference composite; + + final Consumer onError; + + final Action onComplete; + + AbstractDisposableAutoRelease( + DisposableContainer composite, + Consumer onError, + Action onComplete + ) { + this.onError = onError; + this.onComplete = onComplete; + this.composite = new AtomicReference<>(composite); + } + + public final void onError(Throwable t) { + if (get() != DisposableHelper.DISPOSED) { + lazySet(DisposableHelper.DISPOSED); + try { + onError.accept(t); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + RxJavaPlugins.onError(new CompositeException(t, e)); + } + } else { + RxJavaPlugins.onError(t); + } + removeSelf(); + } + + public final void onComplete() { + if (get() != DisposableHelper.DISPOSED) { + lazySet(DisposableHelper.DISPOSED); + try { + onComplete.run(); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + RxJavaPlugins.onError(e); + } + } + removeSelf(); + } + + @Override + public final void dispose() { + DisposableHelper.dispose(this); + removeSelf(); + } + + final void removeSelf() { + DisposableContainer c = composite.getAndSet(null); + if (c != null) { + c.delete(this); + } + } + + @Override + public final boolean isDisposed() { + return DisposableHelper.isDisposed(get()); + } + + public final void onSubscribe(Disposable d) { + DisposableHelper.setOnce(this, d); + } + + @Override + public final boolean hasCustomOnError() { + return onError != Functions.ON_ERROR_MISSING; + } + +} diff --git a/src/main/java/io/reactivex/rxjava3/internal/observers/DisposableAutoReleaseMultiObserver.java b/src/main/java/io/reactivex/rxjava3/internal/observers/DisposableAutoReleaseMultiObserver.java new file mode 100644 index 0000000000..4779ff5723 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava3/internal/observers/DisposableAutoReleaseMultiObserver.java @@ -0,0 +1,78 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import io.reactivex.rxjava3.core.*; +import io.reactivex.rxjava3.disposables.DisposableContainer; +import io.reactivex.rxjava3.exceptions.Exceptions; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.disposables.DisposableHelper; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; + +/** + * Wraps lambda callbacks and when the upstream terminates or this (Single | Maybe | Completable) + * observer gets disposed, removes itself from a {@link io.reactivex.rxjava3.disposables.CompositeDisposable}. + *

History: 0.18.0 @ RxJavaExtensions + * @param the element type consumed + * @since 3.1.0 + */ +public final class DisposableAutoReleaseMultiObserver +extends AbstractDisposableAutoRelease +implements SingleObserver, MaybeObserver, CompletableObserver { + + private static final long serialVersionUID = 8924480688481408726L; + + final Consumer onSuccess; + + public DisposableAutoReleaseMultiObserver( + DisposableContainer composite, + Consumer onSuccess, + Consumer onError, + Action onComplete + ) { + super(composite, onError, onComplete); + this.onSuccess = onSuccess; + } + + @Override + public void onSuccess(T t) { + if (get() != DisposableHelper.DISPOSED) { + lazySet(DisposableHelper.DISPOSED); + try { + onSuccess.accept(t); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + RxJavaPlugins.onError(e); + } + } + removeSelf(); + } + +} diff --git a/src/main/java/io/reactivex/rxjava3/internal/observers/DisposableAutoReleaseObserver.java b/src/main/java/io/reactivex/rxjava3/internal/observers/DisposableAutoReleaseObserver.java new file mode 100644 index 0000000000..89435c96ed --- /dev/null +++ b/src/main/java/io/reactivex/rxjava3/internal/observers/DisposableAutoReleaseObserver.java @@ -0,0 +1,76 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import io.reactivex.rxjava3.core.Observer; +import io.reactivex.rxjava3.disposables.DisposableContainer; +import io.reactivex.rxjava3.exceptions.Exceptions; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.disposables.DisposableHelper; + +/** + * Wraps lambda callbacks and when the upstream terminates or this observer gets disposed, + * removes itself from a {@link io.reactivex.rxjava3.disposables.CompositeDisposable}. + *

History: 0.18.0 @ RxJavaExtensions + * @param the element type consumed + * @since 3.1.0 + */ +public final class DisposableAutoReleaseObserver +extends AbstractDisposableAutoRelease +implements Observer { + + private static final long serialVersionUID = 8924480688481408726L; + + final Consumer onNext; + + public DisposableAutoReleaseObserver( + DisposableContainer composite, + Consumer onNext, + Consumer onError, + Action onComplete + ) { + super(composite, onError, onComplete); + this.onNext = onNext; + } + + @Override + public void onNext(T t) { + if (get() != DisposableHelper.DISPOSED) { + try { + onNext.accept(t); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + get().dispose(); + onError(e); + } + } + } + +} diff --git a/src/main/java/io/reactivex/rxjava3/internal/subscribers/DisposableAutoReleaseSubscriber.java b/src/main/java/io/reactivex/rxjava3/internal/subscribers/DisposableAutoReleaseSubscriber.java new file mode 100644 index 0000000000..3b9d2d99b3 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava3/internal/subscribers/DisposableAutoReleaseSubscriber.java @@ -0,0 +1,151 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.subscribers; + +import java.util.concurrent.atomic.AtomicReference; + +import org.reactivestreams.Subscription; + +import io.reactivex.rxjava3.core.FlowableSubscriber; +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.*; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.functions.Functions; +import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; + +/** + * Wraps lambda callbacks and when the upstream terminates or this subscriber gets disposed, + * removes itself from a {@link io.reactivex.rxjava3.disposables.CompositeDisposable}. + *

History: 0.18.0 @ RxJavaExtensions + * @param the element type consumed + * @since 3.1.0 + */ +public final class DisposableAutoReleaseSubscriber +extends AtomicReference +implements FlowableSubscriber, Disposable, LambdaConsumerIntrospection { + + private static final long serialVersionUID = 8924480688481408726L; + + final AtomicReference composite; + + final Consumer onNext; + + final Consumer onError; + + final Action onComplete; + + public DisposableAutoReleaseSubscriber( + DisposableContainer composite, + Consumer onNext, + Consumer onError, + Action onComplete + ) { + this.onNext = onNext; + this.onError = onError; + this.onComplete = onComplete; + this.composite = new AtomicReference<>(composite); + } + + @Override + public void onNext(T t) { + if (get() != SubscriptionHelper.CANCELLED) { + try { + onNext.accept(t); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + get().cancel(); + onError(e); + } + } + } + + @Override + public void onError(Throwable t) { + if (get() != SubscriptionHelper.CANCELLED) { + lazySet(SubscriptionHelper.CANCELLED); + try { + onError.accept(t); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + RxJavaPlugins.onError(new CompositeException(t, e)); + } + } else { + RxJavaPlugins.onError(t); + } + removeSelf(); + } + + @Override + public void onComplete() { + if (get() != SubscriptionHelper.CANCELLED) { + lazySet(SubscriptionHelper.CANCELLED); + try { + onComplete.run(); + } catch (Throwable e) { + Exceptions.throwIfFatal(e); + RxJavaPlugins.onError(e); + } + } + removeSelf(); + } + + @Override + public void dispose() { + SubscriptionHelper.cancel(this); + removeSelf(); + } + + void removeSelf() { + DisposableContainer c = composite.getAndSet(null); + if (c != null) { + c.delete(this); + } + } + + @Override + public boolean isDisposed() { + return SubscriptionHelper.CANCELLED == get(); + } + + @Override + public void onSubscribe(Subscription s) { + if (SubscriptionHelper.setOnce(this, s)) { + s.request(Long.MAX_VALUE); + } + } + + @Override + public boolean hasCustomOnError() { + return onError != Functions.ON_ERROR_MISSING; + } + +} diff --git a/src/test/java/io/reactivex/rxjava3/internal/observers/CompletableConsumersTest.java b/src/test/java/io/reactivex/rxjava3/internal/observers/CompletableConsumersTest.java new file mode 100644 index 0000000000..eb17ead3db --- /dev/null +++ b/src/test/java/io/reactivex/rxjava3/internal/observers/CompletableConsumersTest.java @@ -0,0 +1,221 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.*; + +import org.junit.Test; + +import io.reactivex.rxjava3.core.*; +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.CompositeException; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.subjects.CompletableSubject; +import io.reactivex.rxjava3.testsupport.TestHelper; + +public class CompletableConsumersTest implements Consumer, Action { + + final CompositeDisposable composite = new CompositeDisposable(); + + final CompletableSubject processor = CompletableSubject.create(); + + final List events = new ArrayList<>(); + + @Override + public void run() throws Exception { + events.add("OnComplete"); + } + + @Override + public void accept(Object t) throws Exception { + events.add(t); + } + + @Test + public void onErrorNormal() { + + processor.subscribe(this, this, composite); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onComplete(); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList("OnComplete"), events); + + } + + @Test + public void onErrorError() { + + Disposable d = processor.subscribe(this, this, composite); + + assertTrue(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onError(new IOException()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteNormal() { + + processor.subscribe(this, this, composite); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onComplete(); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList("OnComplete"), events); + + } + + @Test + public void onCompleteError() { + + processor.subscribe(this, this, composite); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onError(new IOException()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteDispose() { + + Disposable d = processor.subscribe(this, this, composite); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + assertFalse(d.isDisposed()); + + d.dispose(); + d.dispose(); + + assertTrue(d.isDisposed()); + + assertEquals(0, composite.size()); + + assertFalse(processor.hasObservers()); + } + + @Test + public void onErrorCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + processor.subscribe(this, t -> { + throw new IOException(t); + }, composite); + + processor.onError(new IllegalArgumentException()); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, CompositeException.class); + List inners = TestHelper.compositeList(errors.get(0)); + TestHelper.assertError(inners, 0, IllegalArgumentException.class); + TestHelper.assertError(inners, 1, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onCompleteCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + processor.subscribe(new Action() { + @Override + public void run() throws Exception { + throw new IOException(); + } + }, this, composite); + + processor.onComplete(); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + new Completable() { + @Override + protected void subscribeActual( + CompletableObserver observer) { + observer.onSubscribe(Disposable.empty()); + observer.onComplete(); + + observer.onSubscribe(Disposable.empty()); + observer.onComplete(); + observer.onError(new IOException()); + } + }.subscribe(this, this, composite); + + assertEquals(Arrays.asList("OnComplete"), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava3/internal/observers/MaybeConsumersTest.java b/src/test/java/io/reactivex/rxjava3/internal/observers/MaybeConsumersTest.java new file mode 100644 index 0000000000..92b32c000e --- /dev/null +++ b/src/test/java/io/reactivex/rxjava3/internal/observers/MaybeConsumersTest.java @@ -0,0 +1,273 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.*; + +import org.junit.Test; + +import io.reactivex.rxjava3.core.*; +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.CompositeException; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.functions.Functions; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.subjects.MaybeSubject; +import io.reactivex.rxjava3.testsupport.TestHelper; + +public class MaybeConsumersTest implements Consumer, Action { + + final CompositeDisposable composite = new CompositeDisposable(); + + final MaybeSubject processor = MaybeSubject.create(); + + final List events = new ArrayList<>(); + + @Override + public void run() throws Exception { + events.add("OnComplete"); + } + + @Override + public void accept(Object t) throws Exception { + events.add(t); + } + + static Disposable subscribeAutoDispose(Maybe source, CompositeDisposable composite, + Consumer onSuccess, Consumer onError, Action onComplete) { + return source.subscribe(onSuccess, onError, onComplete, composite); + } + + @Test + public void onSuccessNormal() { + + Disposable d = subscribeAutoDispose(processor, composite, this, Functions.ON_ERROR_MISSING, () -> { }); + + assertFalse(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onSuccess(1); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList(1), events); + + } + + @Test + public void onErrorNormal() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onSuccess(1); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList(1), events); + + } + + @Test + public void onErrorError() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onError(new IOException()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteNormal() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onComplete(); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList("OnComplete"), events); + + } + + @Test + public void onCompleteError() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onError(new IOException()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteDispose() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + assertFalse(d.isDisposed()); + + d.dispose(); + d.dispose(); + + assertTrue(d.isDisposed()); + + assertEquals(0, composite.size()); + + assertFalse(processor.hasObservers()); + } + + @Test + public void onSuccessCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, new Consumer() { + @Override + public void accept(Object t) throws Exception { + throw new IOException(); + } + }, this, this); + + processor.onSuccess(1); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onErrorCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, new Consumer() { + @Override + public void accept(Throwable t) throws Exception { + throw new IOException(t); + } + }, this); + + processor.onError(new IllegalArgumentException()); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, CompositeException.class); + List inners = TestHelper.compositeList(errors.get(0)); + TestHelper.assertError(inners, 0, IllegalArgumentException.class); + TestHelper.assertError(inners, 1, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onCompleteCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, this, new Action() { + @Override + public void run() throws Exception { + throw new IOException(); + } + }); + + processor.onComplete(); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose( + new Maybe() { + @Override + protected void subscribeActual( + MaybeObserver observer) { + observer.onSubscribe(Disposable.empty()); + observer.onComplete(); + + observer.onSubscribe(Disposable.empty()); + observer.onSuccess(2); + observer.onComplete(); + observer.onError(new IOException()); + } + }, composite, this, this, this + ); + + assertEquals(Arrays.asList("OnComplete"), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava3/internal/observers/ObservableConsumersTest.java b/src/test/java/io/reactivex/rxjava3/internal/observers/ObservableConsumersTest.java new file mode 100644 index 0000000000..b5d26340e7 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava3/internal/observers/ObservableConsumersTest.java @@ -0,0 +1,324 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.*; + +import org.junit.Test; + +import io.reactivex.rxjava3.core.Observable; +import io.reactivex.rxjava3.core.Observer; +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.*; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.functions.Functions; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.subjects.PublishSubject; +import io.reactivex.rxjava3.testsupport.TestHelper; + +public class ObservableConsumersTest implements Consumer, Action { + + final CompositeDisposable composite = new CompositeDisposable(); + + final PublishSubject processor = PublishSubject.create(); + + final List events = new ArrayList<>(); + + @Override + public void run() throws Exception { + events.add("OnComplete"); + } + + @Override + public void accept(Object t) throws Exception { + events.add(t); + } + + static Disposable subscribeAutoDispose(Observable source, CompositeDisposable composite, + Consumer onNext, Consumer onError, Action onComplete) { + return source.subscribe(onNext, onError, onComplete, composite); + } + + @Test + public void onNextNormal() { + + Disposable d = subscribeAutoDispose(processor, composite, this, Functions.ON_ERROR_MISSING, () -> { }); + + assertFalse(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onComplete(); + + assertEquals(Arrays.asList(1), events); + + assertEquals(0, composite.size()); + } + + @Test + public void onErrorNormal() { + + subscribeAutoDispose(processor, composite, this, Functions.ON_ERROR_MISSING, () -> { }); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onComplete(); + + assertEquals(Arrays.asList(1), events); + + assertEquals(0, composite.size()); + } + + @Test + public void onErrorError() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onError(new IOException()); + + assertEquals(events.toString(), 1, events.get(0)); + assertTrue(events.toString(), events.get(1) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteNormal() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onComplete(); + + assertEquals(Arrays.asList(1, "OnComplete"), events); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteError() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onError(new IOException()); + + assertEquals(events.toString(), 1, events.get(0)); + assertTrue(events.toString(), events.get(1) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteDispose() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + assertFalse(d.isDisposed()); + + d.dispose(); + d.dispose(); + + assertTrue(d.isDisposed()); + + assertEquals(0, composite.size()); + + assertFalse(processor.hasObservers()); + } + + @Test + public void onNextCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, new Consumer() { + @Override + public void accept(Object t) throws Exception { + throw new IOException(); + } + }, this, this); + + processor.onNext(1); + + assertTrue(errors.toString(), errors.isEmpty()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onNextCrashOnError() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, new Consumer() { + @Override + public void accept(Throwable t) throws Exception { + throw new IOException(t); + } + }, this); + + processor.onError(new IllegalArgumentException()); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, CompositeException.class); + List inners = TestHelper.compositeList(errors.get(0)); + TestHelper.assertError(inners, 0, IllegalArgumentException.class); + TestHelper.assertError(inners, 1, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onNextCrashNoError() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, t -> { + throw new IOException(); + }, Functions.ON_ERROR_MISSING, () -> { }); + + processor.onNext(1); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, OnErrorNotImplementedException.class); + assertTrue(errors.get(0).getCause() instanceof IOException); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onCompleteCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, this, new Action() { + @Override + public void run() throws Exception { + throw new IOException(); + } + }); + + processor.onNext(1); + processor.onComplete(); + + assertEquals(Arrays.asList(1), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose( + new Observable() { + @Override + protected void subscribeActual( + Observer observer) { + observer.onSubscribe(Disposable.empty()); + observer.onNext(1); + observer.onComplete(); + + observer.onSubscribe(Disposable.empty()); + observer.onNext(2); + observer.onComplete(); + observer.onError(new IOException()); + } + }, composite, this, this, this + ); + + assertEquals(Arrays.asList(1, "OnComplete"), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava3/internal/observers/SingleConsumersTest.java b/src/test/java/io/reactivex/rxjava3/internal/observers/SingleConsumersTest.java new file mode 100644 index 0000000000..a840d8f71a --- /dev/null +++ b/src/test/java/io/reactivex/rxjava3/internal/observers/SingleConsumersTest.java @@ -0,0 +1,192 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.observers; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.*; + +import org.junit.Test; + +import io.reactivex.rxjava3.core.*; +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.CompositeException; +import io.reactivex.rxjava3.functions.Consumer; +import io.reactivex.rxjava3.internal.functions.Functions; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.subjects.SingleSubject; +import io.reactivex.rxjava3.testsupport.TestHelper; + +public class SingleConsumersTest implements Consumer { + + final CompositeDisposable composite = new CompositeDisposable(); + + final SingleSubject processor = SingleSubject.create(); + + final List events = new ArrayList<>(); + + @Override + public void accept(Object t) throws Exception { + events.add(t); + } + + static Disposable subscribeAutoDispose(Single source, CompositeDisposable composite, + Consumer onSuccess, Consumer onError) { + return source.subscribe(onSuccess, onError, composite); + } + + @Test + public void onSuccessNormal() { + + Disposable d = subscribeAutoDispose(processor, composite, this, Functions.ON_ERROR_MISSING); + + assertFalse(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onSuccess(1); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList(1), events); + + } + + @Test + public void onErrorNormal() { + + subscribeAutoDispose(processor, composite, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onSuccess(1); + + assertEquals(0, composite.size()); + + assertEquals(Arrays.asList(1), events); + + } + + @Test + public void onErrorError() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this); + + assertTrue(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onError(new IOException()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onSuccessCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, new Consumer() { + @Override + public void accept(Object t) throws Exception { + throw new IOException(); + } + }, this); + + processor.onSuccess(1); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onErrorCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, new Consumer() { + @Override + public void accept(Throwable t) throws Exception { + throw new IOException(t); + } + }); + + processor.onError(new IllegalArgumentException()); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, CompositeException.class); + List inners = TestHelper.compositeList(errors.get(0)); + TestHelper.assertError(inners, 0, IllegalArgumentException.class); + TestHelper.assertError(inners, 1, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose( + new Single() { + @Override + protected void subscribeActual( + SingleObserver observer) { + observer.onSubscribe(Disposable.empty()); + observer.onSuccess(1); + + observer.onSubscribe(Disposable.empty()); + observer.onSuccess(2); + observer.onError(new IOException()); + } + }, composite, this, this + ); + + assertEquals(Arrays.asList(1), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava3/internal/subscribers/FlowableConsumersTest.java b/src/test/java/io/reactivex/rxjava3/internal/subscribers/FlowableConsumersTest.java new file mode 100644 index 0000000000..cc7083cde7 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava3/internal/subscribers/FlowableConsumersTest.java @@ -0,0 +1,328 @@ +/* + * 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. + */ + +/* + * Copyright 2016-2019 David Karnok + * + * 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.subscribers; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.*; + +import org.junit.Test; +import org.reactivestreams.Subscriber; + +import io.reactivex.rxjava3.core.*; +import io.reactivex.rxjava3.disposables.*; +import io.reactivex.rxjava3.exceptions.*; +import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.functions.Functions; +import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription; +import io.reactivex.rxjava3.observers.LambdaConsumerIntrospection; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.processors.PublishProcessor; +import io.reactivex.rxjava3.testsupport.TestHelper; + +public class FlowableConsumersTest implements Consumer, Action { + + final CompositeDisposable composite = new CompositeDisposable(); + + final PublishProcessor processor = PublishProcessor.create(); + + final List events = new ArrayList<>(); + + @Override + public void run() throws Exception { + events.add("OnComplete"); + } + + @Override + public void accept(Object t) throws Exception { + events.add(t); + } + + static Disposable subscribeAutoDispose(Flowable source, CompositeDisposable composite, + Consumer onNext, Consumer onError, Action onComplete) { + return source.subscribe(onNext, onError, onComplete, composite); + } + + @Test + public void onNextNormal() { + + Disposable d = subscribeAutoDispose(processor, composite, this, Functions.ON_ERROR_MISSING, () -> { }); + + assertFalse(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onComplete(); + + assertEquals(Arrays.asList(1), events); + + assertEquals(0, composite.size()); + } + + @Test + public void onErrorNormal() { + + subscribeAutoDispose(processor, composite, this, this, () -> { }); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onComplete(); + + assertEquals(Arrays.asList(1), events); + + assertEquals(0, composite.size()); + } + + @Test + public void onErrorError() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(d.getClass().toString(), ((LambdaConsumerIntrospection)d).hasCustomOnError()); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onError(new IOException()); + + assertEquals(events.toString(), 1, events.get(0)); + assertTrue(events.toString(), events.get(1) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteNormal() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onComplete(); + + assertEquals(Arrays.asList(1, "OnComplete"), events); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteError() { + + subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + processor.onNext(1); + + assertTrue(composite.size() > 0); + + assertEquals(Arrays.asList(1), events); + + processor.onError(new IOException()); + + assertEquals(events.toString(), 1, events.get(0)); + assertTrue(events.toString(), events.get(1) instanceof IOException); + + assertEquals(0, composite.size()); + } + + @Test + public void onCompleteDispose() { + + Disposable d = subscribeAutoDispose(processor, composite, this, this, this); + + assertTrue(composite.size() > 0); + + assertTrue(events.toString(), events.isEmpty()); + + assertFalse(d.isDisposed()); + + d.dispose(); + d.dispose(); + + assertTrue(d.isDisposed()); + + assertEquals(0, composite.size()); + + assertFalse(processor.hasSubscribers()); + } + + @Test + public void onNextCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, new Consumer() { + @Override + public void accept(Object t) throws Exception { + throw new IOException(); + } + }, this, this); + + processor.onNext(1); + + assertTrue(errors.toString(), errors.isEmpty()); + + assertTrue(events.toString(), events.get(0) instanceof IOException); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onNextCrashOnError() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, new Consumer() { + @Override + public void accept(Throwable t) throws Exception { + throw new IOException(t); + } + }, this); + + processor.onError(new IllegalArgumentException()); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, CompositeException.class); + List inners = TestHelper.compositeList(errors.get(0)); + TestHelper.assertError(inners, 0, IllegalArgumentException.class); + TestHelper.assertError(inners, 1, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onNextCrashNoError() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, new Consumer() { + @Override + public void accept(Object t) throws Exception { + throw new IOException(); + } + }, Functions.ON_ERROR_MISSING, () -> { }); + + processor.onNext(1); + + assertTrue(events.toString(), events.isEmpty()); + + TestHelper.assertError(errors, 0, OnErrorNotImplementedException.class); + assertTrue(errors.get(0).getCause() instanceof IOException); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void onCompleteCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose(processor, composite, this, this, new Action() { + @Override + public void run() throws Exception { + throw new IOException(); + } + }); + + processor.onNext(1); + processor.onComplete(); + + assertEquals(Arrays.asList(1), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + subscribeAutoDispose( + new Flowable() { + @Override + protected void subscribeActual( + Subscriber s) { + s.onSubscribe(new BooleanSubscription()); + s.onNext(1); + s.onComplete(); + + s.onSubscribe(new BooleanSubscription()); + s.onNext(2); + s.onComplete(); + s.onError(new IOException()); + } + }, composite, this, this, this + ); + + assertEquals(Arrays.asList(1, "OnComplete"), events); + + TestHelper.assertUndeliverable(errors, 0, IOException.class); + } finally { + RxJavaPlugins.reset(); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava3/validators/BaseTypeAnnotations.java b/src/test/java/io/reactivex/rxjava3/validators/BaseTypeAnnotations.java index d03c0dcade..582df0fa71 100644 --- a/src/test/java/io/reactivex/rxjava3/validators/BaseTypeAnnotations.java +++ b/src/test/java/io/reactivex/rxjava3/validators/BaseTypeAnnotations.java @@ -22,6 +22,7 @@ import io.reactivex.rxjava3.annotations.*; import io.reactivex.rxjava3.core.*; +import io.reactivex.rxjava3.disposables.DisposableContainer; import io.reactivex.rxjava3.flowables.ConnectableFlowable; import io.reactivex.rxjava3.observables.ConnectableObservable; import io.reactivex.rxjava3.parallel.ParallelFlowable; @@ -44,7 +45,8 @@ static void checkCheckReturnValueSupport(Class clazz) { for (Method m : clazz.getMethods()) { if (m.getDeclaringClass() == clazz) { - boolean isSubscribeMethod = "subscribe".equals(m.getName()) && m.getParameterTypes().length == 0; + boolean isSubscribeMethod = "subscribe".equals(m.getName()) && + (m.getParameterTypes().length == 0 || m.getParameterTypes()[m.getParameterCount() - 1] == DisposableContainer.class); boolean isConnectMethod = "connect".equals(m.getName()) && m.getParameterTypes().length == 0; boolean isAnnotationPresent = m.isAnnotationPresent(CheckReturnValue.class); diff --git a/src/test/java/io/reactivex/rxjava3/validators/JavadocForAnnotations.java b/src/test/java/io/reactivex/rxjava3/validators/JavadocForAnnotations.java index c3c814f9ff..1c683783d6 100644 --- a/src/test/java/io/reactivex/rxjava3/validators/JavadocForAnnotations.java +++ b/src/test/java/io/reactivex/rxjava3/validators/JavadocForAnnotations.java @@ -95,7 +95,7 @@ static final void scanFor(StringBuilder sourceCode, String annotation, String in ; int lc = lineNumber(sourceCode, idx); - e.append(" at io.reactivex.").append(baseClassName) + e.append(" at io.reactivex.rxjava3.core.").append(baseClassName) .append(" (").append(baseClassName).append(".java:") .append(lc).append(")").append("\r\n\r\n"); } diff --git a/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java b/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java index b09303643c..3ab41589a3 100644 --- a/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java +++ b/src/test/java/io/reactivex/rxjava3/validators/JavadocWording.java @@ -309,6 +309,7 @@ public void flowableDocRefersToFlowableTypes() throws Exception { && !m.signature.contains("Maybe") && !m.signature.contains("MaybeSource") && !m.signature.contains("Disposable") + && !m.signature.contains("void subscribe") ) { CharSequence subSequence = m.javadoc.subSequence(idx - 6, idx + 11); if (idx < 6 || !subSequence.equals("{@link Disposable")) { diff --git a/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java b/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java index 84f99707f2..51c7626ed5 100644 --- a/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java +++ b/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java @@ -25,7 +25,7 @@ import io.reactivex.rxjava3.core.*; import io.reactivex.rxjava3.core.Observable; import io.reactivex.rxjava3.core.Observer; -import io.reactivex.rxjava3.disposables.Disposable; +import io.reactivex.rxjava3.disposables.*; import io.reactivex.rxjava3.exceptions.TestException; import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.Functions; @@ -580,6 +580,8 @@ public void checkParallelFlowable() { defaultValues.put(ParallelFailureHandling.class, ParallelFailureHandling.ERROR); + defaultValues.put(DisposableContainer.class, new CompositeDisposable()); + // JDK 8 types defaultValues.put(Optional.class, Optional.of(1));