Skip to content

Commit 3aae12e

Browse files
akarnokdakarnokd
authored
2.x: Fix Observable.flatMap scalar maxConcurrency overflow (#5900)
1 parent 2edea6b commit 3aae12e

File tree

3 files changed

+157
-12
lines changed

3 files changed

+157
-12
lines changed

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

Lines changed: 22 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -143,16 +143,19 @@ public void onNext(T t) {
143143
void subscribeInner(ObservableSource<? extends U> p) {
144144
for (;;) {
145145
if (p instanceof Callable) {
146-
tryEmitScalar(((Callable<? extends U>)p));
147-
148-
if (maxConcurrency != Integer.MAX_VALUE) {
146+
if (tryEmitScalar(((Callable<? extends U>)p)) && maxConcurrency != Integer.MAX_VALUE) {
147+
boolean empty = false;
149148
synchronized (this) {
150149
p = sources.poll();
151150
if (p == null) {
152151
wip--;
153-
break;
152+
empty = true;
154153
}
155154
}
155+
if (empty) {
156+
drain();
157+
break;
158+
}
156159
} else {
157160
break;
158161
}
@@ -214,26 +217,26 @@ void removeInner(InnerObserver<T, U> inner) {
214217
}
215218
}
216219

217-
void tryEmitScalar(Callable<? extends U> value) {
220+
boolean tryEmitScalar(Callable<? extends U> value) {
218221
U u;
219222
try {
220223
u = value.call();
221224
} catch (Throwable ex) {
222225
Exceptions.throwIfFatal(ex);
223226
errors.addThrowable(ex);
224227
drain();
225-
return;
228+
return true;
226229
}
227230

228231
if (u == null) {
229-
return;
232+
return true;
230233
}
231234

232235

233236
if (get() == 0 && compareAndSet(0, 1)) {
234237
actual.onNext(u);
235238
if (decrementAndGet() == 0) {
236-
return;
239+
return true;
237240
}
238241
} else {
239242
SimplePlainQueue<U> q = queue;
@@ -248,13 +251,14 @@ void tryEmitScalar(Callable<? extends U> value) {
248251

249252
if (!q.offer(u)) {
250253
onError(new IllegalStateException("Scalar queue full?!"));
251-
return;
254+
return true;
252255
}
253256
if (getAndIncrement() != 0) {
254-
return;
257+
return false;
255258
}
256259
}
257260
drainLoop();
261+
return true;
258262
}
259263

260264
void tryEmit(U value, InnerObserver<T, U> inner) {
@@ -360,7 +364,14 @@ void drainLoop() {
360364
InnerObserver<?, ?>[] inner = observers.get();
361365
int n = inner.length;
362366

363-
if (d && (svq == null || svq.isEmpty()) && n == 0) {
367+
int nSources = 0;
368+
if (maxConcurrency != Integer.MAX_VALUE) {
369+
synchronized (this) {
370+
nSources = sources.size();
371+
}
372+
}
373+
374+
if (d && (svq == null || svq.isEmpty()) && n == 0 && nSources == 0) {
364375
Throwable ex = errors.terminate();
365376
if (ex != ExceptionHelper.TERMINATED) {
366377
if (ex == null) {

‎src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapTest.java‎

Lines changed: 68 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222

2323
import io.reactivex.*;
2424
import io.reactivex.disposables.*;
25-
import io.reactivex.exceptions.TestException;
25+
import io.reactivex.exceptions.*;
2626
import io.reactivex.functions.Function;
2727
import io.reactivex.internal.functions.Functions;
2828
import io.reactivex.observers.TestObserver;
@@ -431,4 +431,71 @@ public void onComplete() {
431431

432432
assertTrue(disposable[0].isDisposed());
433433
}
434+
435+
@Test
436+
public void reentrantNoOverflow() {
437+
List<Throwable> errors = TestHelper.trackPluginErrors();
438+
try {
439+
final PublishSubject<Integer> ps = PublishSubject.create();
440+
441+
TestObserver<Integer> to = ps.concatMap(new Function<Integer, Observable<Integer>>() {
442+
@Override
443+
public Observable<Integer> apply(Integer v)
444+
throws Exception {
445+
return Observable.just(v + 1);
446+
}
447+
}, 1)
448+
.subscribeWith(new TestObserver<Integer>() {
449+
@Override
450+
public void onNext(Integer t) {
451+
super.onNext(t);
452+
if (t == 1) {
453+
for (int i = 1; i < 10; i++) {
454+
ps.onNext(i);
455+
}
456+
ps.onComplete();
457+
}
458+
}
459+
});
460+
461+
ps.onNext(0);
462+
463+
if (!errors.isEmpty()) {
464+
to.onError(new CompositeException(errors));
465+
}
466+
467+
to.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
468+
} finally {
469+
RxJavaPlugins.reset();
470+
}
471+
}
472+
473+
@Test
474+
public void reentrantNoOverflowHidden() {
475+
final PublishSubject<Integer> ps = PublishSubject.create();
476+
477+
TestObserver<Integer> to = ps.concatMap(new Function<Integer, Observable<Integer>>() {
478+
@Override
479+
public Observable<Integer> apply(Integer v)
480+
throws Exception {
481+
return Observable.just(v + 1).hide();
482+
}
483+
}, 1)
484+
.subscribeWith(new TestObserver<Integer>() {
485+
@Override
486+
public void onNext(Integer t) {
487+
super.onNext(t);
488+
if (t == 1) {
489+
for (int i = 1; i < 10; i++) {
490+
ps.onNext(i);
491+
}
492+
ps.onComplete();
493+
}
494+
}
495+
});
496+
497+
ps.onNext(0);
498+
499+
to.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
500+
}
434501
}

‎src/test/java/io/reactivex/internal/operators/observable/ObservableFlatMapTest.java‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -938,4 +938,71 @@ public void remove() {
938938

939939
assertEquals(1, counter.get());
940940
}
941+
942+
@Test
943+
public void scalarQueueNoOverflow() {
944+
List<Throwable> errors = TestHelper.trackPluginErrors();
945+
try {
946+
final PublishSubject<Integer> ps = PublishSubject.create();
947+
948+
TestObserver<Integer> to = ps.flatMap(new Function<Integer, Observable<Integer>>() {
949+
@Override
950+
public Observable<Integer> apply(Integer v)
951+
throws Exception {
952+
return Observable.just(v + 1);
953+
}
954+
}, 1)
955+
.subscribeWith(new TestObserver<Integer>() {
956+
@Override
957+
public void onNext(Integer t) {
958+
super.onNext(t);
959+
if (t == 1) {
960+
for (int i = 1; i < 10; i++) {
961+
ps.onNext(i);
962+
}
963+
ps.onComplete();
964+
}
965+
}
966+
});
967+
968+
ps.onNext(0);
969+
970+
if (!errors.isEmpty()) {
971+
to.onError(new CompositeException(errors));
972+
}
973+
974+
to.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
975+
} finally {
976+
RxJavaPlugins.reset();
977+
}
978+
}
979+
980+
@Test
981+
public void scalarQueueNoOverflowHidden() {
982+
final PublishSubject<Integer> ps = PublishSubject.create();
983+
984+
TestObserver<Integer> to = ps.flatMap(new Function<Integer, Observable<Integer>>() {
985+
@Override
986+
public Observable<Integer> apply(Integer v)
987+
throws Exception {
988+
return Observable.just(v + 1).hide();
989+
}
990+
}, 1)
991+
.subscribeWith(new TestObserver<Integer>() {
992+
@Override
993+
public void onNext(Integer t) {
994+
super.onNext(t);
995+
if (t == 1) {
996+
for (int i = 1; i < 10; i++) {
997+
ps.onNext(i);
998+
}
999+
ps.onComplete();
1000+
}
1001+
}
1002+
});
1003+
1004+
ps.onNext(0);
1005+
1006+
to.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
1007+
}
9411008
}

0 commit comments

Comments
 (0)