Skip to content

Commit 4257ef5

Browse files
akarnokdakarnokd
authored
3.x: Add doOnLifecycle to M/S/C (#6877)
1 parent db0bd71 commit 4257ef5

File tree

9 files changed

+817
-20
lines changed

9 files changed

+817
-20
lines changed

‎docs/Operator-Matrix.md

+14-20
Large diffs are not rendered by default.

‎src/main/java/io/reactivex/rxjava3/core/Completable.java

+29
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import org.reactivestreams.*;
1919

2020
import io.reactivex.rxjava3.annotations.*;
21+
import io.reactivex.rxjava3.core.Observable;
2122
import io.reactivex.rxjava3.disposables.Disposable;
2223
import io.reactivex.rxjava3.exceptions.*;
2324
import io.reactivex.rxjava3.functions.*;
@@ -1751,6 +1752,34 @@ public final Completable doOnEvent(@NonNull Consumer<@Nullable ? super Throwable
17511752
return RxJavaPlugins.onAssembly(new CompletableDoOnEvent(this, onEvent));
17521753
}
17531754

1755+
/**
1756+
* Calls the appropriate {@code onXXX} method (shared between all {@link CompletableObserver}s) for the lifecycle events of
1757+
* the sequence (subscription, disposal).
1758+
* <p>
1759+
* <img width="640" height="257" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Completable.doOnLifecycle.png" alt="">
1760+
* <dl>
1761+
* <dt><b>Scheduler:</b></dt>
1762+
* <dd>{@code doOnLifecycle} does not operate by default on a particular {@link Scheduler}.</dd>
1763+
* </dl>
1764+
*
1765+
* @param onSubscribe
1766+
* a {@link Consumer} called with the {@link Disposable} sent via {@link CompletableObserver#onSubscribe(Disposable)}
1767+
* @param onDispose
1768+
* called when the downstream disposes the {@code Disposable} via {@code dispose()}
1769+
* @return the new {@code Completable} instance
1770+
* @throws NullPointerException if {@code onSubscribe} or {@code onDispose} is {@code null}
1771+
* @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
1772+
* @since 3.0.0
1773+
*/
1774+
@CheckReturnValue
1775+
@SchedulerSupport(SchedulerSupport.NONE)
1776+
@NonNull
1777+
public final Completable doOnLifecycle(@NonNull Consumer<? super Disposable> onSubscribe, @NonNull Action onDispose) {
1778+
return doOnLifecycle(onSubscribe, Functions.emptyConsumer(),
1779+
Functions.EMPTY_ACTION, Functions.EMPTY_ACTION,
1780+
Functions.EMPTY_ACTION, onDispose);
1781+
}
1782+
17541783
/**
17551784
* Returns a {@code Completable} instance that calls the various callbacks upon the specific
17561785
* lifecycle events.

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

+28
Original file line numberDiff line numberDiff line change
@@ -3388,6 +3388,34 @@ public final Maybe<T> doOnEvent(@NonNull BiConsumer<@Nullable ? super T, @Nullab
33883388
return RxJavaPlugins.onAssembly(new MaybeDoOnEvent<>(this, onEvent));
33893389
}
33903390

3391+
/**
3392+
* Calls the appropriate {@code onXXX} method (shared between all {@link MaybeObserver}s) for the lifecycle events of
3393+
* the sequence (subscription, disposal).
3394+
* <p>
3395+
* <img width="640" height="183" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.doOnLifecycle.png" alt="">
3396+
* <dl>
3397+
* <dt><b>Scheduler:</b></dt>
3398+
* <dd>{@code doOnLifecycle} does not operate by default on a particular {@link Scheduler}.</dd>
3399+
* </dl>
3400+
*
3401+
* @param onSubscribe
3402+
* a {@link Consumer} called with the {@link Disposable} sent via {@link MaybeObserver#onSubscribe(Disposable)}
3403+
* @param onDispose
3404+
* called when the downstream disposes the {@code Disposable} via {@code dispose()}
3405+
* @return the new {@code Maybe} instance
3406+
* @throws NullPointerException if {@code onSubscribe} or {@code onDispose} is {@code null}
3407+
* @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
3408+
* @since 3.0.0
3409+
*/
3410+
@CheckReturnValue
3411+
@SchedulerSupport(SchedulerSupport.NONE)
3412+
@NonNull
3413+
public final Maybe<T> doOnLifecycle(@NonNull Consumer<? super Disposable> onSubscribe, @NonNull Action onDispose) {
3414+
Objects.requireNonNull(onSubscribe, "onSubscribe is null");
3415+
Objects.requireNonNull(onDispose, "onDispose is null");
3416+
return RxJavaPlugins.onAssembly(new MaybeDoOnLifecycle<>(this, onSubscribe, onDispose));
3417+
}
3418+
33913419
/**
33923420
* Calls the shared {@link Consumer} with the {@link Disposable} sent through the {@code onSubscribe} for each
33933421
* {@link MaybeObserver} that subscribes to the current {@code Maybe}.

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

+29
Original file line numberDiff line numberDiff line change
@@ -2684,6 +2684,34 @@ public final Single<T> doFinally(@NonNull Action onFinally) {
26842684
return RxJavaPlugins.onAssembly(new SingleDoFinally<>(this, onFinally));
26852685
}
26862686

2687+
/**
2688+
* Calls the appropriate {@code onXXX} method (shared between all {@link SingleObserver}s) for the lifecycle events of
2689+
* the sequence (subscription, disposal).
2690+
* <p>
2691+
* <img width="640" height="232" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.doOnLifecycle.png" alt="">
2692+
* <dl>
2693+
* <dt><b>Scheduler:</b></dt>
2694+
* <dd>{@code doOnLifecycle} does not operate by default on a particular {@link Scheduler}.</dd>
2695+
* </dl>
2696+
*
2697+
* @param onSubscribe
2698+
* a {@link Consumer} called with the {@link Disposable} sent via {@link SingleObserver#onSubscribe(Disposable)}
2699+
* @param onDispose
2700+
* called when the downstream disposes the {@code Disposable} via {@code dispose()}
2701+
* @return the new {@code Single} instance
2702+
* @throws NullPointerException if {@code onSubscribe} or {@code onDispose} is {@code null}
2703+
* @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
2704+
* @since 3.0.0
2705+
*/
2706+
@CheckReturnValue
2707+
@SchedulerSupport(SchedulerSupport.NONE)
2708+
@NonNull
2709+
public final Single<T> doOnLifecycle(@NonNull Consumer<? super Disposable> onSubscribe, @NonNull Action onDispose) {
2710+
Objects.requireNonNull(onSubscribe, "onSubscribe is null");
2711+
Objects.requireNonNull(onDispose, "onDispose is null");
2712+
return RxJavaPlugins.onAssembly(new SingleDoOnLifecycle<>(this, onSubscribe, onDispose));
2713+
}
2714+
26872715
/**
26882716
* Calls the shared consumer with the {@link Disposable} sent through the {@code onSubscribe} for each
26892717
* {@link SingleObserver} that subscribes to the current {@code Single}.
@@ -3455,6 +3483,7 @@ public final Flowable<T> mergeWith(@NonNull SingleSource<? extends T> other) {
34553483
* @return the new {@link Maybe} instance
34563484
* @throws NullPointerException if {@code clazz} is {@code null}
34573485
* @see <a href="http://reactivex.io/documentation/operators/filter.html">ReactiveX operators documentation: Filter</a>
3486+
* @since 3.0.0
34583487
*/
34593488
@CheckReturnValue
34603489
@NonNull

‎src/main/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeDoOnLifecycle.java

+125
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
package io.reactivex.rxjava3.internal.operators.maybe;
14+
15+
import io.reactivex.rxjava3.annotations.NonNull;
16+
import io.reactivex.rxjava3.core.*;
17+
import io.reactivex.rxjava3.disposables.Disposable;
18+
import io.reactivex.rxjava3.exceptions.Exceptions;
19+
import io.reactivex.rxjava3.functions.*;
20+
import io.reactivex.rxjava3.internal.disposables.*;
21+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
22+
23+
/**
24+
* Invokes callbacks upon {@code onSubscribe} from upstream and
25+
* {@code dispose} from downstream.
26+
*
27+
* @param <T> the element type of the flow
28+
* @since 3.0.0
29+
*/
30+
public final class MaybeDoOnLifecycle<T> extends AbstractMaybeWithUpstream<T, T> {
31+
32+
final Consumer<? super Disposable> onSubscribe;
33+
34+
final Action onDispose;
35+
36+
public MaybeDoOnLifecycle(Maybe<T> upstream, Consumer<? super Disposable> onSubscribe,
37+
Action onDispose) {
38+
super(upstream);
39+
this.onSubscribe = onSubscribe;
40+
this.onDispose = onDispose;
41+
}
42+
43+
@Override
44+
protected void subscribeActual(MaybeObserver<? super T> observer) {
45+
source.subscribe(new MaybeLifecycleObserver<>(observer, onSubscribe, onDispose));
46+
}
47+
48+
static final class MaybeLifecycleObserver<T> implements MaybeObserver<T>, Disposable {
49+
50+
final MaybeObserver<? super T> downstream;
51+
52+
final Consumer<? super Disposable> onSubscribe;
53+
54+
final Action onDispose;
55+
56+
Disposable upstream;
57+
58+
MaybeLifecycleObserver(MaybeObserver<? super T> downstream, Consumer<? super Disposable> onSubscribe, Action onDispose) {
59+
this.downstream = downstream;
60+
this.onSubscribe = onSubscribe;
61+
this.onDispose = onDispose;
62+
}
63+
64+
@Override
65+
public void onSubscribe(@NonNull Disposable d) {
66+
// this way, multiple calls to onSubscribe can show up in tests that use doOnSubscribe to validate behavior
67+
try {
68+
onSubscribe.accept(d);
69+
} catch (Throwable e) {
70+
Exceptions.throwIfFatal(e);
71+
d.dispose();
72+
this.upstream = DisposableHelper.DISPOSED;
73+
EmptyDisposable.error(e, downstream);
74+
return;
75+
}
76+
if (DisposableHelper.validate(this.upstream, d)) {
77+
this.upstream = d;
78+
downstream.onSubscribe(this);
79+
}
80+
}
81+
82+
@Override
83+
public void onSuccess(@NonNull T t) {
84+
if (upstream != DisposableHelper.DISPOSED) {
85+
upstream = DisposableHelper.DISPOSED;
86+
downstream.onSuccess(t);
87+
}
88+
}
89+
90+
@Override
91+
public void onError(@NonNull Throwable e) {
92+
if (upstream != DisposableHelper.DISPOSED) {
93+
upstream = DisposableHelper.DISPOSED;
94+
downstream.onError(e);
95+
} else {
96+
RxJavaPlugins.onError(e);
97+
}
98+
}
99+
100+
@Override
101+
public void onComplete() {
102+
if (upstream != DisposableHelper.DISPOSED) {
103+
upstream = DisposableHelper.DISPOSED;
104+
downstream.onComplete();
105+
}
106+
}
107+
108+
@Override
109+
public void dispose() {
110+
try {
111+
onDispose.run();
112+
} catch (Throwable e) {
113+
Exceptions.throwIfFatal(e);
114+
RxJavaPlugins.onError(e);
115+
}
116+
upstream.dispose();
117+
upstream = DisposableHelper.DISPOSED;
118+
}
119+
120+
@Override
121+
public boolean isDisposed() {
122+
return upstream.isDisposed();
123+
}
124+
}
125+
}

‎src/main/java/io/reactivex/rxjava3/internal/operators/single/SingleDoOnLifecycle.java

+119
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
package io.reactivex.rxjava3.internal.operators.single;
14+
15+
import io.reactivex.rxjava3.annotations.NonNull;
16+
import io.reactivex.rxjava3.core.*;
17+
import io.reactivex.rxjava3.disposables.Disposable;
18+
import io.reactivex.rxjava3.exceptions.Exceptions;
19+
import io.reactivex.rxjava3.functions.*;
20+
import io.reactivex.rxjava3.internal.disposables.*;
21+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
22+
23+
/**
24+
* Invokes callbacks upon {@code onSubscribe} from upstream and
25+
* {@code dispose} from downstream.
26+
*
27+
* @param <T> the element type of the flow
28+
* @since 3.0.0
29+
*/
30+
public final class SingleDoOnLifecycle<T> extends Single<T> {
31+
32+
final Single<T> source;
33+
34+
final Consumer<? super Disposable> onSubscribe;
35+
36+
final Action onDispose;
37+
38+
public SingleDoOnLifecycle(Single<T> upstream, Consumer<? super Disposable> onSubscribe,
39+
Action onDispose) {
40+
this.source = upstream;
41+
this.onSubscribe = onSubscribe;
42+
this.onDispose = onDispose;
43+
}
44+
45+
@Override
46+
protected void subscribeActual(SingleObserver<? super T> observer) {
47+
source.subscribe(new SingleLifecycleObserver<>(observer, onSubscribe, onDispose));
48+
}
49+
50+
static final class SingleLifecycleObserver<T> implements SingleObserver<T>, Disposable {
51+
52+
final SingleObserver<? super T> downstream;
53+
54+
final Consumer<? super Disposable> onSubscribe;
55+
56+
final Action onDispose;
57+
58+
Disposable upstream;
59+
60+
SingleLifecycleObserver(SingleObserver<? super T> downstream, Consumer<? super Disposable> onSubscribe, Action onDispose) {
61+
this.downstream = downstream;
62+
this.onSubscribe = onSubscribe;
63+
this.onDispose = onDispose;
64+
}
65+
66+
@Override
67+
public void onSubscribe(@NonNull Disposable d) {
68+
// this way, multiple calls to onSubscribe can show up in tests that use doOnSubscribe to validate behavior
69+
try {
70+
onSubscribe.accept(d);
71+
} catch (Throwable e) {
72+
Exceptions.throwIfFatal(e);
73+
d.dispose();
74+
this.upstream = DisposableHelper.DISPOSED;
75+
EmptyDisposable.error(e, downstream);
76+
return;
77+
}
78+
if (DisposableHelper.validate(this.upstream, d)) {
79+
this.upstream = d;
80+
downstream.onSubscribe(this);
81+
}
82+
}
83+
84+
@Override
85+
public void onSuccess(@NonNull T t) {
86+
if (upstream != DisposableHelper.DISPOSED) {
87+
upstream = DisposableHelper.DISPOSED;
88+
downstream.onSuccess(t);
89+
}
90+
}
91+
92+
@Override
93+
public void onError(@NonNull Throwable e) {
94+
if (upstream != DisposableHelper.DISPOSED) {
95+
upstream = DisposableHelper.DISPOSED;
96+
downstream.onError(e);
97+
} else {
98+
RxJavaPlugins.onError(e);
99+
}
100+
}
101+
102+
@Override
103+
public void dispose() {
104+
try {
105+
onDispose.run();
106+
} catch (Throwable e) {
107+
Exceptions.throwIfFatal(e);
108+
RxJavaPlugins.onError(e);
109+
}
110+
upstream.dispose();
111+
upstream = DisposableHelper.DISPOSED;
112+
}
113+
114+
@Override
115+
public boolean isDisposed() {
116+
return upstream.isDisposed();
117+
}
118+
}
119+
}

0 commit comments

Comments
 (0)