Skip to content

Commit 43ceedf

Browse files
akarnokdakarnokd
authored
2.x: Dedicated {Single|Maybe}.flatMap{Publisher|Observable} & andThen(Observable|Publisher) implementations (#6024)
* 2.x: Dedicated {0..1}.flatMap{Publisher|Obs} & andThen implementations * Fix local variable name.
1 parent 07586f4 commit 43ceedf

15 files changed

+1086
-18
lines changed

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

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,8 @@
2424
import io.reactivex.internal.fuseable.*;
2525
import io.reactivex.internal.observers.*;
2626
import io.reactivex.internal.operators.completable.*;
27-
import io.reactivex.internal.operators.flowable.FlowableDelaySubscriptionOther;
2827
import io.reactivex.internal.operators.maybe.*;
29-
import io.reactivex.internal.operators.observable.ObservableDelaySubscriptionOther;
28+
import io.reactivex.internal.operators.mixed.*;
3029
import io.reactivex.internal.operators.single.SingleDelayWithCompletable;
3130
import io.reactivex.internal.util.ExceptionHelper;
3231
import io.reactivex.observers.TestObserver;
@@ -872,7 +871,7 @@ public final Completable ambWith(CompletableSource other) {
872871
@SchedulerSupport(SchedulerSupport.NONE)
873872
public final <T> Observable<T> andThen(ObservableSource<T> next) {
874873
ObjectHelper.requireNonNull(next, "next is null");
875-
return RxJavaPlugins.onAssembly(new ObservableDelaySubscriptionOther<T, Object>(next, toObservable()));
874+
return RxJavaPlugins.onAssembly(new CompletableAndThenObservable<T>(this, next));
876875
}
877876

878877
/**
@@ -897,7 +896,7 @@ public final <T> Observable<T> andThen(ObservableSource<T> next) {
897896
@SchedulerSupport(SchedulerSupport.NONE)
898897
public final <T> Flowable<T> andThen(Publisher<T> next) {
899898
ObjectHelper.requireNonNull(next, "next is null");
900-
return RxJavaPlugins.onAssembly(new FlowableDelaySubscriptionOther<T, Object>(next, toFlowable()));
899+
return RxJavaPlugins.onAssembly(new CompletableAndThenPublisher<T>(this, next));
901900
}
902901

903902
/**

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import io.reactivex.internal.observers.BlockingMultiObserver;
2828
import io.reactivex.internal.operators.flowable.*;
2929
import io.reactivex.internal.operators.maybe.*;
30+
import io.reactivex.internal.operators.mixed.*;
3031
import io.reactivex.internal.util.*;
3132
import io.reactivex.observers.TestObserver;
3233
import io.reactivex.plugins.RxJavaPlugins;
@@ -2961,7 +2962,8 @@ public final <U> Observable<U> flattenAsObservable(final Function<? super T, ? e
29612962
@CheckReturnValue
29622963
@SchedulerSupport(SchedulerSupport.NONE)
29632964
public final <R> Observable<R> flatMapObservable(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
2964-
return toObservable().flatMap(mapper);
2965+
ObjectHelper.requireNonNull(mapper, "mapper is null");
2966+
return RxJavaPlugins.onAssembly(new MaybeFlatMapObservable<T, R>(this, mapper));
29652967
}
29662968

29672969
/**
@@ -2987,7 +2989,8 @@ public final <R> Observable<R> flatMapObservable(Function<? super T, ? extends O
29872989
@CheckReturnValue
29882990
@SchedulerSupport(SchedulerSupport.NONE)
29892991
public final <R> Flowable<R> flatMapPublisher(Function<? super T, ? extends Publisher<? extends R>> mapper) {
2990-
return toFlowable().flatMap(mapper);
2992+
ObjectHelper.requireNonNull(mapper, "mapper is null");
2993+
return RxJavaPlugins.onAssembly(new MaybeFlatMapPublisher<T, R>(this, mapper));
29912994
}
29922995

29932996
/**

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import io.reactivex.internal.operators.completable.*;
2929
import io.reactivex.internal.operators.flowable.*;
3030
import io.reactivex.internal.operators.maybe.*;
31+
import io.reactivex.internal.operators.mixed.*;
3132
import io.reactivex.internal.operators.observable.*;
3233
import io.reactivex.internal.operators.single.*;
3334
import io.reactivex.internal.util.*;
@@ -2535,7 +2536,8 @@ public final <U> Observable<U> flattenAsObservable(final Function<? super T, ? e
25352536
@CheckReturnValue
25362537
@SchedulerSupport(SchedulerSupport.NONE)
25372538
public final <R> Observable<R> flatMapObservable(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
2538-
return toObservable().flatMap(mapper);
2539+
ObjectHelper.requireNonNull(mapper, "mapper is null");
2540+
return RxJavaPlugins.onAssembly(new SingleFlatMapObservable<T, R>(this, mapper));
25392541
}
25402542

25412543
/**

‎src/main/java/io/reactivex/internal/operators/mixed/CompletableAndThenObservable.java

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
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+
14+
package io.reactivex.internal.operators.mixed;
15+
16+
import java.util.concurrent.atomic.AtomicReference;
17+
18+
import io.reactivex.*;
19+
import io.reactivex.disposables.Disposable;
20+
import io.reactivex.internal.disposables.DisposableHelper;
21+
22+
/**
23+
* After Completable completes, it relays the signals
24+
* of the ObservableSource to the downstream observer.
25+
*
26+
* @param <R> the result type of the ObservableSource and this operator
27+
* @since 2.1.15
28+
*/
29+
public final class CompletableAndThenObservable<R> extends Observable<R> {
30+
31+
final CompletableSource source;
32+
33+
final ObservableSource<? extends R> other;
34+
35+
public CompletableAndThenObservable(CompletableSource source,
36+
ObservableSource<? extends R> other) {
37+
this.source = source;
38+
this.other = other;
39+
}
40+
41+
@Override
42+
protected void subscribeActual(Observer<? super R> s) {
43+
AndThenObservableObserver<R> parent = new AndThenObservableObserver<R>(s, other);
44+
s.onSubscribe(parent);
45+
source.subscribe(parent);
46+
}
47+
48+
static final class AndThenObservableObserver<R>
49+
extends AtomicReference<Disposable>
50+
implements Observer<R>, CompletableObserver, Disposable {
51+
52+
private static final long serialVersionUID = -8948264376121066672L;
53+
54+
final Observer<? super R> downstream;
55+
56+
ObservableSource<? extends R> other;
57+
58+
AndThenObservableObserver(Observer<? super R> downstream, ObservableSource<? extends R> other) {
59+
this.other = other;
60+
this.downstream = downstream;
61+
}
62+
63+
@Override
64+
public void onNext(R t) {
65+
downstream.onNext(t);
66+
}
67+
68+
@Override
69+
public void onError(Throwable t) {
70+
downstream.onError(t);
71+
}
72+
73+
@Override
74+
public void onComplete() {
75+
ObservableSource<? extends R> o = other;
76+
if (o == null) {
77+
downstream.onComplete();
78+
} else {
79+
other = null;
80+
o.subscribe(this);
81+
}
82+
}
83+
84+
85+
@Override
86+
public void dispose() {
87+
DisposableHelper.dispose(this);
88+
}
89+
90+
@Override
91+
public boolean isDisposed() {
92+
return DisposableHelper.isDisposed(get());
93+
}
94+
95+
@Override
96+
public void onSubscribe(Disposable d) {
97+
DisposableHelper.replace(this, d);
98+
}
99+
100+
}
101+
}

‎src/main/java/io/reactivex/internal/operators/mixed/CompletableAndThenPublisher.java

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
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+
14+
package io.reactivex.internal.operators.mixed;
15+
16+
import java.util.concurrent.atomic.*;
17+
18+
import org.reactivestreams.*;
19+
20+
import io.reactivex.*;
21+
import io.reactivex.disposables.Disposable;
22+
import io.reactivex.internal.disposables.DisposableHelper;
23+
import io.reactivex.internal.subscriptions.SubscriptionHelper;
24+
25+
/**
26+
* After Completable completes, it relays the signals
27+
* of the Publisher to the downstream subscriber.
28+
*
29+
* @param <R> the result type of the Publisher and this operator
30+
* @since 2.1.15
31+
*/
32+
public final class CompletableAndThenPublisher<R> extends Flowable<R> {
33+
34+
final CompletableSource source;
35+
36+
final Publisher<? extends R> other;
37+
38+
public CompletableAndThenPublisher(CompletableSource source,
39+
Publisher<? extends R> other) {
40+
this.source = source;
41+
this.other = other;
42+
}
43+
44+
@Override
45+
protected void subscribeActual(Subscriber<? super R> s) {
46+
source.subscribe(new AndThenPublisherSubscriber<R>(s, other));
47+
}
48+
49+
static final class AndThenPublisherSubscriber<R>
50+
extends AtomicReference<Subscription>
51+
implements FlowableSubscriber<R>, CompletableObserver, Subscription {
52+
53+
private static final long serialVersionUID = -8948264376121066672L;
54+
55+
final Subscriber<? super R> downstream;
56+
57+
Publisher<? extends R> other;
58+
59+
Disposable upstream;
60+
61+
final AtomicLong requested;
62+
63+
AndThenPublisherSubscriber(Subscriber<? super R> downstream, Publisher<? extends R> other) {
64+
this.downstream = downstream;
65+
this.other = other;
66+
this.requested = new AtomicLong();
67+
}
68+
69+
@Override
70+
public void onNext(R t) {
71+
downstream.onNext(t);
72+
}
73+
74+
@Override
75+
public void onError(Throwable t) {
76+
downstream.onError(t);
77+
}
78+
79+
@Override
80+
public void onComplete() {
81+
Publisher<? extends R> p = other;
82+
if (p == null) {
83+
downstream.onComplete();
84+
} else {
85+
other = null;
86+
p.subscribe(this);
87+
}
88+
}
89+
90+
@Override
91+
public void request(long n) {
92+
SubscriptionHelper.deferredRequest(this, requested, n);
93+
}
94+
95+
@Override
96+
public void cancel() {
97+
upstream.dispose();
98+
SubscriptionHelper.cancel(this);
99+
}
100+
101+
@Override
102+
public void onSubscribe(Disposable d) {
103+
if (DisposableHelper.validate(upstream, d)) {
104+
this.upstream = d;
105+
downstream.onSubscribe(this);
106+
}
107+
}
108+
109+
@Override
110+
public void onSubscribe(Subscription s) {
111+
SubscriptionHelper.deferredSetOnce(this, requested, s);
112+
}
113+
}
114+
}

0 commit comments

Comments
 (0)