Skip to content

Commit bbae4a5

Browse files
akarnokdakarnokd
authored
2.x: add doAfterNext & doAfterSuccess to the other types (#4835)
1 parent 715d71d commit bbae4a5

File tree

9 files changed

+887
-0
lines changed

9 files changed

+887
-0
lines changed

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2273,6 +2273,25 @@ public final Maybe<T> delaySubscription(long delay, TimeUnit unit, Scheduler sch
22732273
return delaySubscription(Flowable.timer(delay, unit, scheduler));
22742274
}
22752275

2276+
/**
2277+
* Calls the specified consumer with the success item after this item has been emitted to the downstream.
2278+
* <p>Note that the {@code onAfterNext} action is shared between subscriptions and as such
2279+
* should be thread-safe.
2280+
* <dl>
2281+
* <dt><b>Scheduler:</b></dt>
2282+
* <dd>{@code doAfterSuccess} does not operate by default on a particular {@link Scheduler}.</dd>
2283+
* </dl>
2284+
* @param onAfterSuccess the Consumer that will be called after emitting an item from upstream to the downstream
2285+
* @return the new Maybe instance
2286+
* @since 2.0.1 - experimental
2287+
*/
2288+
@SchedulerSupport(SchedulerSupport.NONE)
2289+
@Experimental
2290+
public final Maybe<T> doAfterSuccess(Consumer<? super T> onAfterSuccess) {
2291+
ObjectHelper.requireNonNull(onAfterSuccess, "doAfterSuccess is null");
2292+
return RxJavaPlugins.onAssembly(new MaybeDoAfterSuccess<T>(this, onAfterSuccess));
2293+
}
2294+
22762295
/**
22772296
* Registers an {@link Action} to be called when this Maybe invokes either
22782297
* {@link MaybeObserver#onComplete onSuccess},

‎src/main/java/io/reactivex/Observable.java

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6416,6 +6416,27 @@ public final Observable<T> distinctUntilChanged(BiPredicate<? super T, ? super T
64166416
return RxJavaPlugins.onAssembly(new ObservableDistinctUntilChanged<T, T>(this, Functions.<T>identity(), comparer));
64176417
}
64186418

6419+
/**
6420+
* Calls the specified consumer with the current item after this item has been emitted to the downstream.
6421+
* <p>Note that the {@code onAfterNext} action is shared between subscriptions and as such
6422+
* should be thread-safe.
6423+
* <dl>
6424+
* <dt><b>Scheduler:</b></dt>
6425+
* <dd>{@code doAfterNext} does not operate by default on a particular {@link Scheduler}.</dd>
6426+
* <td><b>Operator-fusion:</b></dt>
6427+
* <dd>This operator supports boundary-limited synchronous or asynchronous queue-fusion.</dd>
6428+
* </dl>
6429+
* @param onAfterNext the Consumer that will be called after emitting an item from upstream to the downstream
6430+
* @return the new Observable instance
6431+
* @since 2.0.1 - experimental
6432+
*/
6433+
@SchedulerSupport(SchedulerSupport.NONE)
6434+
@Experimental
6435+
public final Observable<T> doAfterNext(Consumer<? super T> onAfterNext) {
6436+
ObjectHelper.requireNonNull(onAfterNext, "onAfterNext is null");
6437+
return RxJavaPlugins.onAssembly(new ObservableDoAfterNext<T>(this, onAfterNext));
6438+
}
6439+
64196440
/**
64206441
* Registers an {@link Action} to be called when this ObservableSource invokes either
64216442
* {@link Observer#onComplete onComplete} or {@link Observer#onError onError}.

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1716,6 +1716,25 @@ public final <U> Single<T> delaySubscription(long time, TimeUnit unit, Scheduler
17161716
return delaySubscription(Observable.timer(time, unit, scheduler));
17171717
}
17181718

1719+
/**
1720+
* Calls the specified consumer with the success item after this item has been emitted to the downstream.
1721+
* <p>Note that the {@code doAfterSuccess} action is shared between subscriptions and as such
1722+
* should be thread-safe.
1723+
* <dl>
1724+
* <dt><b>Scheduler:</b></dt>
1725+
* <dd>{@code doAfterSuccess} does not operate by default on a particular {@link Scheduler}.</dd>
1726+
* </dl>
1727+
* @param onAfterSuccess the Consumer that will be called after emitting an item from upstream to the downstream
1728+
* @return the new Single instance
1729+
* @since 2.0.1 - experimental
1730+
*/
1731+
@SchedulerSupport(SchedulerSupport.NONE)
1732+
@Experimental
1733+
public final Single<T> doAfterSuccess(Consumer<? super T> onAfterSuccess) {
1734+
ObjectHelper.requireNonNull(onAfterSuccess, "doAfterSuccess is null");
1735+
return RxJavaPlugins.onAssembly(new SingleDoAfterSuccess<T>(this, onAfterSuccess));
1736+
}
1737+
17191738
/**
17201739
* Calls the specified action after this Single signals onSuccess or onError or gets disposed by
17211740
* the downstream.

‎src/main/java/io/reactivex/internal/operators/maybe/MaybeDoAfterSuccess.java

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
/**
2+
* Copyright 2016 Netflix, Inc.
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+
14+
package io.reactivex.internal.operators.maybe;
15+
16+
import io.reactivex.*;
17+
import io.reactivex.annotations.Experimental;
18+
import io.reactivex.disposables.Disposable;
19+
import io.reactivex.exceptions.Exceptions;
20+
import io.reactivex.functions.Consumer;
21+
import io.reactivex.internal.disposables.DisposableHelper;
22+
import io.reactivex.plugins.RxJavaPlugins;
23+
24+
/**
25+
* Calls a consumer after pushing the current item to the downstream.
26+
* @param <T> the value type
27+
* @since 2.0.1 - experimental
28+
*/
29+
@Experimental
30+
public final class MaybeDoAfterSuccess<T> extends AbstractMaybeWithUpstream<T, T> {
31+
32+
final Consumer<? super T> onAfterSuccess;
33+
34+
public MaybeDoAfterSuccess(MaybeSource<T> source, Consumer<? super T> onAfterSuccess) {
35+
super(source);
36+
this.onAfterSuccess = onAfterSuccess;
37+
}
38+
39+
@Override
40+
protected void subscribeActual(MaybeObserver<? super T> s) {
41+
source.subscribe(new DoAfterObserver<T>(s, onAfterSuccess));
42+
}
43+
44+
static final class DoAfterObserver<T> implements MaybeObserver<T>, Disposable {
45+
46+
final MaybeObserver<? super T> actual;
47+
48+
final Consumer<? super T> onAfterSuccess;
49+
50+
Disposable d;
51+
52+
DoAfterObserver(MaybeObserver<? super T> actual, Consumer<? super T> onAfterSuccess) {
53+
this.actual = actual;
54+
this.onAfterSuccess = onAfterSuccess;
55+
}
56+
57+
@Override
58+
public void onSubscribe(Disposable d) {
59+
if (DisposableHelper.validate(this.d, d)) {
60+
this.d = d;
61+
62+
actual.onSubscribe(this);
63+
}
64+
}
65+
66+
@Override
67+
public void onSuccess(T t) {
68+
actual.onSuccess(t);
69+
70+
try {
71+
onAfterSuccess.accept(t);
72+
} catch (Throwable ex) {
73+
Exceptions.throwIfFatal(ex);
74+
// remember, onSuccess is a terminal event and we can't call onError
75+
RxJavaPlugins.onError(ex);
76+
}
77+
}
78+
79+
@Override
80+
public void onError(Throwable e) {
81+
actual.onError(e);
82+
}
83+
84+
@Override
85+
public void onComplete() {
86+
actual.onComplete();
87+
}
88+
89+
@Override
90+
public void dispose() {
91+
d.dispose();
92+
}
93+
94+
@Override
95+
public boolean isDisposed() {
96+
return d.isDisposed();
97+
}
98+
}
99+
}

‎src/main/java/io/reactivex/internal/operators/observable/ObservableDoAfterNext.java

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/**
2+
* Copyright 2016 Netflix, Inc.
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+
14+
package io.reactivex.internal.operators.observable;
15+
16+
import io.reactivex.*;
17+
import io.reactivex.annotations.Experimental;
18+
import io.reactivex.functions.Consumer;
19+
import io.reactivex.internal.observers.BasicFuseableObserver;
20+
21+
/**
22+
* Calls a consumer after pushing the current item to the downstream.
23+
* @param <T> the value type
24+
* @since 2.0.1 - experimental
25+
*/
26+
@Experimental
27+
public final class ObservableDoAfterNext<T> extends AbstractObservableWithUpstream<T, T> {
28+
29+
final Consumer<? super T> onAfterNext;
30+
31+
public ObservableDoAfterNext(ObservableSource<T> source, Consumer<? super T> onAfterNext) {
32+
super(source);
33+
this.onAfterNext = onAfterNext;
34+
}
35+
36+
@Override
37+
protected void subscribeActual(Observer<? super T> s) {
38+
source.subscribe(new DoAfterObserver<T>(s, onAfterNext));
39+
}
40+
41+
static final class DoAfterObserver<T> extends BasicFuseableObserver<T, T> {
42+
43+
final Consumer<? super T> onAfterNext;
44+
45+
DoAfterObserver(Observer<? super T> actual, Consumer<? super T> onAfterNext) {
46+
super(actual);
47+
this.onAfterNext = onAfterNext;
48+
}
49+
50+
@Override
51+
public void onNext(T t) {
52+
actual.onNext(t);
53+
54+
if (sourceMode == NONE) {
55+
try {
56+
onAfterNext.accept(t);
57+
} catch (Throwable ex) {
58+
fail(ex);
59+
}
60+
}
61+
}
62+
63+
@Override
64+
public int requestFusion(int mode) {
65+
return transitiveBoundaryFusion(mode);
66+
}
67+
68+
@Override
69+
public T poll() throws Exception {
70+
T v = qs.poll();
71+
if (v != null) {
72+
onAfterNext.accept(v);
73+
}
74+
return v;
75+
}
76+
}
77+
}

‎src/main/java/io/reactivex/internal/operators/single/SingleDoAfterSuccess.java

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/**
2+
* Copyright 2016 Netflix, Inc.
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+
14+
package io.reactivex.internal.operators.single;
15+
16+
import io.reactivex.*;
17+
import io.reactivex.annotations.Experimental;
18+
import io.reactivex.disposables.Disposable;
19+
import io.reactivex.exceptions.Exceptions;
20+
import io.reactivex.functions.Consumer;
21+
import io.reactivex.internal.disposables.DisposableHelper;
22+
import io.reactivex.plugins.RxJavaPlugins;
23+
24+
/**
25+
* Calls a consumer after pushing the current item to the downstream.
26+
* @param <T> the value type
27+
* @since 2.0.1 - experimental
28+
*/
29+
@Experimental
30+
public final class SingleDoAfterSuccess<T> extends Single<T> {
31+
32+
final SingleSource<T> source;
33+
34+
final Consumer<? super T> onAfterSuccess;
35+
36+
public SingleDoAfterSuccess(SingleSource<T> source, Consumer<? super T> onAfterSuccess) {
37+
this.source = source;
38+
this.onAfterSuccess = onAfterSuccess;
39+
}
40+
41+
@Override
42+
protected void subscribeActual(SingleObserver<? super T> s) {
43+
source.subscribe(new DoAfterObserver<T>(s, onAfterSuccess));
44+
}
45+
46+
static final class DoAfterObserver<T> implements SingleObserver<T>, Disposable {
47+
48+
final SingleObserver<? super T> actual;
49+
50+
final Consumer<? super T> onAfterSuccess;
51+
52+
Disposable d;
53+
54+
DoAfterObserver(SingleObserver<? super T> actual, Consumer<? super T> onAfterSuccess) {
55+
this.actual = actual;
56+
this.onAfterSuccess = onAfterSuccess;
57+
}
58+
59+
@Override
60+
public void onSubscribe(Disposable d) {
61+
if (DisposableHelper.validate(this.d, d)) {
62+
this.d = d;
63+
64+
actual.onSubscribe(this);
65+
}
66+
}
67+
68+
@Override
69+
public void onSuccess(T t) {
70+
actual.onSuccess(t);
71+
72+
try {
73+
onAfterSuccess.accept(t);
74+
} catch (Throwable ex) {
75+
Exceptions.throwIfFatal(ex);
76+
// remember, onSuccess is a terminal event and we can't call onError
77+
RxJavaPlugins.onError(ex);
78+
}
79+
}
80+
81+
@Override
82+
public void onError(Throwable e) {
83+
actual.onError(e);
84+
}
85+
86+
@Override
87+
public void dispose() {
88+
d.dispose();
89+
}
90+
91+
@Override
92+
public boolean isDisposed() {
93+
return d.isDisposed();
94+
}
95+
}
96+
}

0 commit comments

Comments
 (0)