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.

Sign up for GitHub

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

Jump to bottom

3.x: Add subscribe with disposable container #7298

Merged
merged 1 commit into from
Jul 17, 2021
Merged

3.x: Add subscribe with disposable container #7298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
45 changes: 43 additions & 2 deletions src/main/java/io/reactivex/rxjava3/core/Completable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.*;
Expand All @@ -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;
Expand Down Expand Up @@ -2836,6 +2836,7 @@ public final Completable hide() {
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
* @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
Expand Down Expand Up @@ -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
Expand All @@ -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.
* <p>
* The {@code CompletableObserver} will be removed after the callback for the terminal event has been invoked.
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
* @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<? super Throwable> onError,
@NonNull DisposableContainer container) {
Objects.requireNonNull(onComplete, "onComplete is null");
Objects.requireNonNull(onError, "onError is null");
Objects.requireNonNull(container, "container is null");

DisposableAutoReleaseMultiObserver<Void> 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.
Expand All @@ -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
Expand Down
53 changes: 51 additions & 2 deletions src/main/java/io/reactivex/rxjava3/core/Flowable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.*;
Expand Down Expand Up @@ -15699,6 +15699,7 @@ public final Flowable<T> startWithArray(@NonNull T... items) {
*
* @return the new {@link Disposable} instance that allows cancelling the flow
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
@SchedulerSupport(SchedulerSupport.NONE)
Expand Down Expand Up @@ -15727,6 +15728,7 @@ public final Disposable subscribe() {
* @throws NullPointerException
* if {@code onNext} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
Expand All @@ -15753,9 +15755,10 @@ public final Disposable subscribe(@NonNull Consumer<? super T> onNext) {
* the {@code Consumer<Throwable>} 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 <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @throws NullPointerException
* if {@code onNext} or {@code onError} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
Expand Down Expand Up @@ -15788,6 +15791,7 @@ public final Disposable subscribe(@NonNull Consumer<? super T> onNext, @NonNull
* @throws NullPointerException
* if {@code onNext}, {@code onError} or {@code onComplete} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
Expand All @@ -15806,6 +15810,51 @@ public final Disposable subscribe(@NonNull Consumer<? super T> 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.
* <p>
* The {@coded Subscriber} will be removed after the callback for the terminal event has been invoked.
* <dl>
* <dt><b>Backpressure:</b></dt>
* <dd>The operator consumes the current {@code Flowable} in an unbounded manner (i.e., no
* backpressure is applied to it).</dd>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
* @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<? super T> onNext,
@NonNull Consumer<? super Throwable> 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<T> subscriber = new DisposableAutoReleaseSubscriber<>(
container, onNext, onError, onComplete);
container.add(subscriber);
subscribe(subscriber);
return subscriber;
}

@BackpressureSupport(BackpressureKind.SPECIAL)
@SchedulerSupport(SchedulerSupport.NONE)
@Override
Expand Down
49 changes: 47 additions & 2 deletions src/main/java/io/reactivex/rxjava3/core/Maybe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.*;
Expand Down Expand Up @@ -5226,6 +5226,7 @@ public final Flowable<T> startWith(@NonNull Publisher<T> other) {
*
* @return the new {@link Disposable} instance that can be used for disposing the subscription at any time
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@SchedulerSupport(SchedulerSupport.NONE)
@NonNull
Expand All @@ -5250,6 +5251,7 @@ public final Disposable subscribe() {
* @throws NullPointerException
* if {@code onSuccess} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
Expand All @@ -5272,10 +5274,11 @@ public final Disposable subscribe(@NonNull Consumer<? super T> onSuccess) {
* the {@code Consumer<Throwable>} 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 <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @throws NullPointerException
* if {@code onSuccess} is {@code null}, or
* if {@code onError} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
Expand Down Expand Up @@ -5305,6 +5308,7 @@ public final Disposable subscribe(@NonNull Consumer<? super T> onSuccess, @NonNu
* if {@code onSuccess}, {@code onError} or
* {@code onComplete} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@NonNull
Expand All @@ -5317,6 +5321,47 @@ public final Disposable subscribe(@NonNull Consumer<? super T> 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.
* <p>
* The {@code MaybeObserver} will be removed after the callback for the terminal event has been invoked.
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
* @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<? super T> onSuccess,
@NonNull Consumer<? super Throwable> 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<T> observer = new DisposableAutoReleaseMultiObserver<>(
container, onSuccess, onError, onComplete);
container.add(observer);
subscribe(observer);
return observer;
}

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(@NonNull MaybeObserver<? super T> observer) {
Expand Down
49 changes: 47 additions & 2 deletions src/main/java/io/reactivex/rxjava3/core/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.*;
Expand Down Expand Up @@ -13025,6 +13025,7 @@ public final Observable<T> startWithArray(@NonNull T... items) {
*
* @return the new {@link Disposable} instance that can be used to dispose the subscription at any time
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@SchedulerSupport(SchedulerSupport.NONE)
@NonNull
Expand All @@ -13049,6 +13050,7 @@ public final Disposable subscribe() {
* @throws NullPointerException
* if {@code onNext} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
Expand All @@ -13071,9 +13073,10 @@ public final Disposable subscribe(@NonNull Consumer<? super T> onNext) {
* the {@code Consumer<Throwable>} 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 <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @throws NullPointerException
* if {@code onNext} or {@code onError} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
Expand Down Expand Up @@ -13102,6 +13105,7 @@ public final Disposable subscribe(@NonNull Consumer<? super T> onNext, @NonNull
* @throws NullPointerException
* if {@code onNext}, {@code onError} or {@code onComplete} is {@code null}
* @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
* @see #subscribe(Consumer, Consumer, Action, DisposableContainer)
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
Expand All @@ -13119,6 +13123,47 @@ public final Disposable subscribe(@NonNull Consumer<? super T> 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.
* <p>
* The {@code Observer} will be removed after the callback for the terminal event has been invoked.
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code subscribe} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
* @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<? super T> onNext,
@NonNull Consumer<? super Throwable> 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<T> observer = new DisposableAutoReleaseObserver<>(
container, onNext, onError, onComplete);
container.add(observer);
subscribe(observer);
return observer;
}

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(@NonNull Observer<? super T> observer) {
Expand Down
Loading