Skip to content

Commit c110f69

Browse files
akarnokdakarnokd
authored
1.x: update map() and filter() to implement OnSubscribe directly (ReactiveX#4097)
1 parent c60a08b commit c110f69

File tree

8 files changed

+120
-35
lines changed

8 files changed

+120
-35
lines changed

‎src/main/java/rx/Observable.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5816,7 +5816,7 @@ public final Observable<Boolean> exists(Func1<? super T, Boolean> predicate) {
58165816
* @see <a href="http://reactivex.io/documentation/operators/filter.html">ReactiveX operators documentation: Filter</a>
58175817
*/
58185818
public final Observable<T> filter(Func1<? super T, Boolean> predicate) {
5819-
return lift(new OperatorFilter<T>(predicate));
5819+
return create(new OnSubscribeFilter<T>(this, predicate));
58205820
}
58215821

58225822
/**
@@ -6623,7 +6623,7 @@ public final Observable<T> limit(int count) {
66236623
* @see <a href="http://reactivex.io/documentation/operators/map.html">ReactiveX operators documentation: Map</a>
66246624
*/
66256625
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
6626-
return lift(new OperatorMap<T, R>(func));
6626+
return create(new OnSubscribeMap<T, R>(this, func));
66276627
}
66286628

66296629
private <R> Observable<R> mapNotification(Func1<? super T, ? extends R> onNext, Func1<? super Throwable, ? extends R> onError, Func0<? extends R> onCompleted) {

‎src/main/java/rx/Single.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1413,7 +1413,7 @@ public final <R> Observable<R> flatMapObservable(Func1<? super T, ? extends Obse
14131413
* @see <a href="http://reactivex.io/documentation/operators/map.html">ReactiveX operators documentation: Map</a>
14141414
*/
14151415
public final <R> Single<R> map(Func1<? super T, ? extends R> func) {
1416-
return lift(new OperatorMap<T, R>(func));
1416+
return create(new SingleOnSubscribeMap<T, R>(this, func));
14171417
}
14181418

14191419
/**

src/main/java/rx/internal/operators/OperatorFilter.java renamed to src/main/java/rx/internal/operators/OnSubscribeFilter.java

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
package rx.internal.operators;
1717

1818
import rx.*;
19-
import rx.Observable.Operator;
19+
import rx.Observable.OnSubscribe;
2020
import rx.exceptions.*;
2121
import rx.functions.Func1;
2222
import rx.internal.util.RxJavaPluginUtils;
@@ -27,19 +27,22 @@
2727
* <img width="640" src="https://github.com/ReactiveX/RxJava/wiki/images/rx-operators/filter.png" alt="">
2828
* @param <T> the value type
2929
*/
30-
public final class OperatorFilter<T> implements Operator<T, T> {
30+
public final class OnSubscribeFilter<T> implements OnSubscribe<T> {
3131

32+
final Observable<T> source;
33+
3234
final Func1<? super T, Boolean> predicate;
3335

34-
public OperatorFilter(Func1<? super T, Boolean> predicate) {
36+
public OnSubscribeFilter(Observable<T> source, Func1<? super T, Boolean> predicate) {
37+
this.source = source;
3538
this.predicate = predicate;
3639
}
3740

3841
@Override
39-
public Subscriber<? super T> call(final Subscriber<? super T> child) {
42+
public void call(final Subscriber<? super T> child) {
4043
FilterSubscriber<T> parent = new FilterSubscriber<T>(child, predicate);
4144
child.add(parent);
42-
return parent;
45+
source.unsafeSubscribe(parent);
4346
}
4447

4548
static final class FilterSubscriber<T> extends Subscriber<T> {

src/main/java/rx/internal/operators/OperatorMap.java renamed to src/main/java/rx/internal/operators/OnSubscribeMap.java

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
package rx.internal.operators;
1717

1818
import rx.*;
19-
import rx.Observable.Operator;
19+
import rx.Observable.OnSubscribe;
2020
import rx.exceptions.*;
2121
import rx.functions.Func1;
2222
import rx.internal.util.RxJavaPluginUtils;
@@ -30,19 +30,22 @@
3030
* @param <T> the input value type
3131
* @param <R> the return value type
3232
*/
33-
public final class OperatorMap<T, R> implements Operator<R, T> {
33+
public final class OnSubscribeMap<T, R> implements OnSubscribe<R> {
3434

35+
final Observable<T> source;
36+
3537
final Func1<? super T, ? extends R> transformer;
3638

37-
public OperatorMap(Func1<? super T, ? extends R> transformer) {
39+
public OnSubscribeMap(Observable<T> source, Func1<? super T, ? extends R> transformer) {
40+
this.source = source;
3841
this.transformer = transformer;
3942
}
40-
43+
4144
@Override
42-
public Subscriber<? super T> call(final Subscriber<? super R> o) {
45+
public void call(final Subscriber<? super R> o) {
4346
MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);
4447
o.add(parent);
45-
return parent;
48+
source.unsafeSubscribe(parent);
4649
}
4750

4851
static final class MapSubscriber<T, R> extends Subscriber<T> {

‎src/main/java/rx/internal/operators/SingleOnSubscribeMap.java‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
/**
2+
* Copyright 2014 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package rx.internal.operators;
17+
18+
import rx.*;
19+
import rx.exceptions.*;
20+
import rx.functions.Func1;
21+
import rx.internal.util.RxJavaPluginUtils;
22+
23+
/**
24+
* Applies a function of your choosing to every item emitted by an {@code Single}, and emits the results of
25+
* this transformation as a new {@code Single}.
26+
*
27+
* @param <T> the input value type
28+
* @param <R> the return value type
29+
*/
30+
public final class SingleOnSubscribeMap<T, R> implements Single.OnSubscribe<R> {
31+
32+
final Single<T> source;
33+
34+
final Func1<? super T, ? extends R> transformer;
35+
36+
public SingleOnSubscribeMap(Single<T> source, Func1<? super T, ? extends R> transformer) {
37+
this.source = source;
38+
this.transformer = transformer;
39+
}
40+
41+
@Override
42+
public void call(final SingleSubscriber<? super R> o) {
43+
MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);
44+
o.add(parent);
45+
source.subscribe(parent);
46+
}
47+
48+
static final class MapSubscriber<T, R> extends SingleSubscriber<T> {
49+
50+
final SingleSubscriber<? super R> actual;
51+
52+
final Func1<? super T, ? extends R> mapper;
53+
54+
boolean done;
55+
56+
public MapSubscriber(SingleSubscriber<? super R> actual, Func1<? super T, ? extends R> mapper) {
57+
this.actual = actual;
58+
this.mapper = mapper;
59+
}
60+
61+
@Override
62+
public void onSuccess(T t) {
63+
R result;
64+
65+
try {
66+
result = mapper.call(t);
67+
} catch (Throwable ex) {
68+
Exceptions.throwIfFatal(ex);
69+
unsubscribe();
70+
onError(OnErrorThrowable.addValueAsLastCause(ex, t));
71+
return;
72+
}
73+
74+
actual.onSuccess(result);
75+
}
76+
77+
@Override
78+
public void onError(Throwable e) {
79+
if (done) {
80+
RxJavaPluginUtils.handleException(e);
81+
return;
82+
}
83+
done = true;
84+
85+
actual.onError(e);
86+
}
87+
}
88+
89+
}
90+

‎src/perf/java/rx/operators/OperatorMapPerf.java‎

Lines changed: 2 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -17,17 +17,9 @@
1717

1818
import java.util.concurrent.TimeUnit;
1919

20-
import org.openjdk.jmh.annotations.BenchmarkMode;
21-
import org.openjdk.jmh.annotations.Benchmark;
22-
import org.openjdk.jmh.annotations.Mode;
23-
import org.openjdk.jmh.annotations.OutputTimeUnit;
24-
import org.openjdk.jmh.annotations.Param;
25-
import org.openjdk.jmh.annotations.Scope;
26-
import org.openjdk.jmh.annotations.State;
20+
import org.openjdk.jmh.annotations.*;
2721

28-
import rx.Observable.Operator;
2922
import rx.functions.Func1;
30-
import rx.internal.operators.OperatorMap;
3123
import rx.jmh.InputWithIncrementingInteger;
3224

3325
@BenchmarkMode(Mode.Throughput)
@@ -49,7 +41,7 @@ public int getSize() {
4941

5042
@Benchmark
5143
public void mapPassThruViaLift(Input input) throws InterruptedException {
52-
input.observable.lift(MAP_OPERATOR).subscribe(input.observer);
44+
input.observable.map(IDENTITY_FUNCTION).subscribe(input.observer);
5345
}
5446

5547
@Benchmark
@@ -63,7 +55,4 @@ public Integer call(Integer value) {
6355
return value;
6456
}
6557
};
66-
67-
private static final Operator<Integer, Integer> MAP_OPERATOR = new OperatorMap<Integer, Integer>(IDENTITY_FUNCTION);
68-
6958
}

‎src/test/java/rx/ObservableConversionTest.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,8 @@
3131
import rx.exceptions.OnErrorNotImplementedException;
3232
import rx.functions.Func1;
3333
import rx.functions.Func2;
34-
import rx.internal.operators.OperatorFilter;
35-
import rx.internal.operators.OperatorMap;
34+
import rx.internal.operators.OnSubscribeFilter;
35+
import rx.internal.operators.OnSubscribeMap;
3636
import rx.observers.TestSubscriber;
3737
import rx.schedulers.Schedulers;
3838

@@ -76,11 +76,11 @@ public <R> CylonDetectorObservable<? extends R> compose(Func1<CylonDetectorObser
7676
}
7777

7878
public final CylonDetectorObservable<T> beep(Func1<? super T, Boolean> predicate) {
79-
return lift(new OperatorFilter<T>(predicate));
79+
return create(new OnSubscribeFilter<T>(Observable.create(onSubscribe), predicate));
8080
}
8181

8282
public final <R> CylonDetectorObservable<R> boop(Func1<? super T, ? extends R> func) {
83-
return lift(new OperatorMap<T, R>(func));
83+
return create(new OnSubscribeMap<T, R>(Observable.create(onSubscribe), func));
8484
}
8585

8686
public CylonDetectorObservable<String> DESTROY() {

‎src/test/java/rx/internal/operators/OperatorMapTest.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -57,14 +57,14 @@ public void testMap() {
5757
Map<String, String> m2 = getMap("Two");
5858
Observable<Map<String, String>> observable = Observable.just(m1, m2);
5959

60-
Observable<String> m = observable.lift(new OperatorMap<Map<String, String>, String>(new Func1<Map<String, String>, String>() {
60+
Observable<String> m = observable.map(new Func1<Map<String, String>, String>() {
6161

6262
@Override
6363
public String call(Map<String, String> map) {
6464
return map.get("firstName");
6565
}
6666

67-
}));
67+
});
6868
m.subscribe(stringObserver);
6969

7070
verify(stringObserver, never()).onError(any(Throwable.class));
@@ -155,15 +155,15 @@ public String call(Map<String, String> map) {
155155
@Test
156156
public void testMapWithError() {
157157
Observable<String> w = Observable.just("one", "fail", "two", "three", "fail");
158-
Observable<String> m = w.lift(new OperatorMap<String, String>(new Func1<String, String>() {
158+
Observable<String> m = w.map(new Func1<String, String>() {
159159
@Override
160160
public String call(String s) {
161161
if ("fail".equals(s)) {
162162
throw new RuntimeException("Forced Failure");
163163
}
164164
return s;
165165
}
166-
})).doOnError(new Action1<Throwable>() {
166+
}).doOnError(new Action1<Throwable>() {
167167

168168
@Override
169169
public void call(Throwable t1) {

0 commit comments

Comments
 (0)