Skip to content

Commit f2ed079

Browse files
akarnokdakarnokd
committed
1.x: optimize concatMapIterable/flatMapIterable (#3864)
* 1.x: optimize concatMapIterable/flatMapIterable * Reformat else if
1 parent 4e5c17c commit f2ed079

File tree

4 files changed

+838
-4
lines changed

4 files changed

+838
-4
lines changed

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4061,7 +4061,7 @@ public final <R> Observable<R> concatMapDelayError(Func1<? super T, ? extends Ob
40614061
* @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
40624062
*/
40634063
public final <R> Observable<R> concatMapIterable(Func1<? super T, ? extends Iterable<? extends R>> collectionSelector) {
4064-
return concat(map(OperatorMapPair.convertSelector(collectionSelector)));
4064+
return OnSubscribeFlattenIterable.createFrom(this, collectionSelector, RxRingBuffer.SIZE);
40654065
}
40664066

40674067
/**
@@ -5672,7 +5672,7 @@ public final <U, R> Observable<R> flatMap(final Func1<? super T, ? extends Obser
56725672
* @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
56735673
*/
56745674
public final <R> Observable<R> flatMapIterable(Func1<? super T, ? extends Iterable<? extends R>> collectionSelector) {
5675-
return merge(map(OperatorMapPair.convertSelector(collectionSelector)));
5675+
return flatMapIterable(collectionSelector, RxRingBuffer.SIZE);
56765676
}
56775677

56785678
/**
@@ -5702,7 +5702,7 @@ public final <R> Observable<R> flatMapIterable(Func1<? super T, ? extends Iterab
57025702
*/
57035703
@Beta
57045704
public final <R> Observable<R> flatMapIterable(Func1<? super T, ? extends Iterable<? extends R>> collectionSelector, int maxConcurrent) {
5705-
return merge(map(OperatorMapPair.convertSelector(collectionSelector)), maxConcurrent);
5705+
return OnSubscribeFlattenIterable.createFrom(this, collectionSelector, maxConcurrent);
57065706
}
57075707

57085708
/**

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

Lines changed: 357 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,357 @@
1+
/**
2+
* Copyright 2016 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+
17+
package rx.internal.operators;
18+
19+
import java.util.*;
20+
import java.util.concurrent.atomic.*;
21+
22+
import rx.*;
23+
import rx.Observable;
24+
import rx.Observable.OnSubscribe;
25+
import rx.exceptions.*;
26+
import rx.functions.Func1;
27+
import rx.internal.util.*;
28+
import rx.internal.util.atomic.*;
29+
import rx.internal.util.unsafe.*;
30+
31+
/**
32+
* Flattens a sequence if Iterable sources, generated via a function, into a single sequence.
33+
*
34+
* @param <T> the input value type
35+
* @param <R> the output value type
36+
*/
37+
public final class OnSubscribeFlattenIterable<T, R> implements OnSubscribe<R> {
38+
39+
final Observable<? extends T> source;
40+
41+
final Func1<? super T, ? extends Iterable<? extends R>> mapper;
42+
43+
final int prefetch;
44+
45+
/** Protected: use createFrom to handle source-dependent optimizations. */
46+
protected OnSubscribeFlattenIterable(Observable<? extends T> source,
47+
Func1<? super T, ? extends Iterable<? extends R>> mapper, int prefetch) {
48+
this.source = source;
49+
this.mapper = mapper;
50+
this.prefetch = prefetch;
51+
}
52+
53+
@Override
54+
public void call(Subscriber<? super R> t) {
55+
final FlattenIterableSubscriber<T, R> parent = new FlattenIterableSubscriber<T, R>(t, mapper, prefetch);
56+
57+
t.add(parent);
58+
t.setProducer(new Producer() {
59+
@Override
60+
public void request(long n) {
61+
parent.requestMore(n);
62+
}
63+
});
64+
65+
source.unsafeSubscribe(parent);
66+
}
67+
68+
public static <T, R> Observable<R> createFrom(Observable<? extends T> source,
69+
Func1<? super T, ? extends Iterable<? extends R>> mapper, int prefetch) {
70+
if (source instanceof ScalarSynchronousObservable) {
71+
T scalar = ((ScalarSynchronousObservable<? extends T>) source).get();
72+
return Observable.create(new OnSubscribeScalarFlattenIterable<T, R>(scalar, mapper));
73+
}
74+
return Observable.create(new OnSubscribeFlattenIterable<T, R>(source, mapper, prefetch));
75+
}
76+
77+
static final class FlattenIterableSubscriber<T, R> extends Subscriber<T> {
78+
final Subscriber<? super R> actual;
79+
80+
final Func1<? super T, ? extends Iterable<? extends R>> mapper;
81+
82+
final long limit;
83+
84+
final Queue<Object> queue;
85+
86+
final AtomicReference<Throwable> error;
87+
88+
final AtomicLong requested;
89+
90+
final AtomicInteger wip;
91+
92+
final NotificationLite<T> nl;
93+
94+
volatile boolean done;
95+
96+
long produced;
97+
98+
Iterator<? extends R> active;
99+
100+
public FlattenIterableSubscriber(Subscriber<? super R> actual,
101+
Func1<? super T, ? extends Iterable<? extends R>> mapper, int prefetch) {
102+
this.actual = actual;
103+
this.mapper = mapper;
104+
this.error = new AtomicReference<Throwable>();
105+
this.wip = new AtomicInteger();
106+
this.requested = new AtomicLong();
107+
this.nl = NotificationLite.instance();
108+
if (prefetch == Integer.MAX_VALUE) {
109+
this.limit = Long.MAX_VALUE;
110+
this.queue = new SpscLinkedArrayQueue<Object>(RxRingBuffer.SIZE);
111+
} else {
112+
// limit = prefetch * 75% rounded up
113+
this.limit = prefetch - (prefetch >> 2);
114+
if (UnsafeAccess.isUnsafeAvailable()) {
115+
this.queue = new SpscArrayQueue<Object>(prefetch);
116+
} else {
117+
this.queue = new SpscAtomicArrayQueue<Object>(prefetch);
118+
}
119+
}
120+
request(prefetch);
121+
}
122+
123+
@Override
124+
public void onNext(T t) {
125+
if (!queue.offer(nl.next(t))) {
126+
unsubscribe();
127+
onError(new MissingBackpressureException());
128+
return;
129+
}
130+
drain();
131+
}
132+
133+
@Override
134+
public void onError(Throwable e) {
135+
if (ExceptionsUtils.addThrowable(error, e)) {
136+
done = true;
137+
drain();
138+
} else {
139+
RxJavaPluginUtils.handleException(e);
140+
}
141+
}
142+
143+
@Override
144+
public void onCompleted() {
145+
done = true;
146+
drain();
147+
}
148+
149+
void requestMore(long n) {
150+
if (n > 0) {
151+
BackpressureUtils.getAndAddRequest(requested, n);
152+
drain();
153+
} else if (n < 0) {
154+
throw new IllegalStateException("n >= 0 required but it was " + n);
155+
}
156+
}
157+
158+
void drain() {
159+
if (wip.getAndIncrement() != 0) {
160+
return;
161+
}
162+
163+
final Subscriber<? super R> actual = this.actual;
164+
final Queue<Object> queue = this.queue;
165+
166+
int missed = 1;
167+
168+
for (;;) {
169+
170+
Iterator<? extends R> it = active;
171+
172+
if (it == null) {
173+
boolean d = done;
174+
175+
Object v = queue.poll();
176+
177+
boolean empty = v == null;
178+
179+
if (checkTerminated(d, empty, actual, queue)) {
180+
return;
181+
}
182+
183+
if (!empty) {
184+
185+
long p = produced + 1;
186+
if (p == limit) {
187+
produced = 0L;
188+
request(p);
189+
} else {
190+
produced = p;
191+
}
192+
193+
boolean b;
194+
195+
try {
196+
Iterable<? extends R> iter = mapper.call(nl.getValue(v));
197+
198+
it = iter.iterator();
199+
200+
b = it.hasNext();
201+
} catch (Throwable ex) {
202+
Exceptions.throwIfFatal(ex);
203+
204+
it = null;
205+
onError(ex);
206+
207+
continue;
208+
}
209+
210+
if (!b) {
211+
continue;
212+
}
213+
214+
active = it;
215+
}
216+
}
217+
218+
if (it != null) {
219+
long r = requested.get();
220+
long e = 0L;
221+
222+
while (e != r) {
223+
if (checkTerminated(done, false, actual, queue)) {
224+
return;
225+
}
226+
227+
R v;
228+
229+
try {
230+
v = it.next();
231+
} catch (Throwable ex) {
232+
Exceptions.throwIfFatal(ex);
233+
it = null;
234+
active = null;
235+
onError(ex);
236+
break;
237+
}
238+
239+
actual.onNext(v);
240+
241+
if (checkTerminated(done, false, actual, queue)) {
242+
return;
243+
}
244+
245+
e++;
246+
247+
boolean b;
248+
249+
try {
250+
b = it.hasNext();
251+
} catch (Throwable ex) {
252+
Exceptions.throwIfFatal(ex);
253+
it = null;
254+
active = null;
255+
onError(ex);
256+
break;
257+
}
258+
259+
if (!b) {
260+
it = null;
261+
active = null;
262+
break;
263+
}
264+
}
265+
266+
if (e == r) {
267+
if (checkTerminated(done, queue.isEmpty() && it == null, actual, queue)) {
268+
return;
269+
}
270+
}
271+
272+
if (e != 0L) {
273+
BackpressureUtils.produced(requested, e);
274+
}
275+
276+
if (it == null) {
277+
continue;
278+
}
279+
}
280+
281+
missed = wip.addAndGet(-missed);
282+
if (missed == 0) {
283+
break;
284+
}
285+
}
286+
}
287+
288+
boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, Queue<?> q) {
289+
if (a.isUnsubscribed()) {
290+
q.clear();
291+
active = null;
292+
return true;
293+
}
294+
295+
if (d) {
296+
Throwable ex = error.get();
297+
if (ex != null) {
298+
ex = ExceptionsUtils.terminate(error);
299+
unsubscribe();
300+
q.clear();
301+
active = null;
302+
303+
a.onError(ex);
304+
return true;
305+
} else
306+
if (empty) {
307+
308+
a.onCompleted();
309+
return true;
310+
}
311+
}
312+
313+
return false;
314+
}
315+
}
316+
317+
/**
318+
* A custom flattener that works from a scalar value and computes the iterable
319+
* during subscription time.
320+
*
321+
* @param <T> the scalar's value type
322+
* @param <R> the result value type
323+
*/
324+
static final class OnSubscribeScalarFlattenIterable<T, R> implements OnSubscribe<R> {
325+
final T value;
326+
327+
final Func1<? super T, ? extends Iterable<? extends R>> mapper;
328+
329+
public OnSubscribeScalarFlattenIterable(T value, Func1<? super T, ? extends Iterable<? extends R>> mapper) {
330+
this.value = value;
331+
this.mapper = mapper;
332+
}
333+
334+
@Override
335+
public void call(Subscriber<? super R> t) {
336+
Iterator<? extends R> itor;
337+
boolean b;
338+
try {
339+
Iterable<? extends R> it = mapper.call(value);
340+
341+
itor = it.iterator();
342+
343+
b = itor.hasNext();
344+
} catch (Throwable ex) {
345+
Exceptions.throwOrReport(ex, t, value);
346+
return;
347+
}
348+
349+
if (!b) {
350+
t.onCompleted();
351+
return;
352+
}
353+
354+
t.setProducer(new OnSubscribeFromIterable.IterableProducer<R>(t, itor));
355+
}
356+
}
357+
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public void call(final Subscriber<? super T> o) {
6464
}
6565
}
6666

67-
private static final class IterableProducer<T> extends AtomicLong implements Producer {
67+
static final class IterableProducer<T> extends AtomicLong implements Producer {
6868
/** */
6969
private static final long serialVersionUID = -8730475647105475802L;
7070
private final Subscriber<? super T> o;

0 commit comments

Comments
 (0)