# FlatMap (part 2)

DevFeed: [FlatMap (part 2)](<https://devfeed.tech/articles/flatmap-part-2-24796.md>)

Original publisher: [Read original article](<https://akarnokd.blogspot.com/2016/03/flatmap-part-2.html>)

Author: David Karnok (noreply@blogger.com)

Published: 2016-03-02T22:50:00Z

Content type: tutorial

Language: en

Sources: [Akarnokd - Advanced RxJava](<https://devfeed.tech/sources/akarnokd-advanced-rxjava.md>)

Topics: [Concurrency](<https://devfeed.tech/topics/concurrency.md>), [implementation](<https://devfeed.tech/topics/implementation.md>), [Network](<https://devfeed.tech/topics/network.md>), [Data structures](<https://devfeed.tech/topics/data-structures.md>)

Tags: [backpressure](<https://devfeed.tech/tags/backpressure.md>), [concurrency](<https://devfeed.tech/tags/concurrency.md>), [concurrent](<https://devfeed.tech/tags/concurrent.md>), [errors](<https://devfeed.tech/tags/errors.md>), [implementation](<https://devfeed.tech/tags/implementation.md>), [network](<https://devfeed.tech/tags/network.md>), [parameter](<https://devfeed.tech/tags/parameter.md>), [performance](<https://devfeed.tech/tags/performance.md>), [queue](<https://devfeed.tech/tags/queue.md>)

## AI overview

This post extends a flatMap implementation with bounded concurrency and delayed error handling. It uses backpressure to limit active inner Observables and collects errors for a final CompositeException.

## Source excerpt

Introduction In this post, we will look into expanding the features of our flatMap implementation and improve its performance. RxJava's flatMap implementation offers limiting the maximum concurrency, that is, the maximum number of active subscriptions to the generated sources and allows delaying exceptions coming from any of the sources, including the main. Limiting concurrency Due to historical reasons, RxJava's flatMap (and our version of it from part 1) is unbounded towards the main source. This may work with infrequent main emissions and/or short lived inner Observable sequences. However, even if the main source, such as range(), can emit at any rate, the mapped inner Observables may consume limited resources such as network connections. So the question is, how can we make sure only an user defined number of active Observables are being merged at once? How can we make sure some source emits only a limited number of values? The answer is, of course, backpressure. To limit the concurrency in flatMap, the idea is to request a maxConcurrency amount upfront via request(), and then whenever a source completes, request(1) extra. Let's change our OpFlatMap and FlatMapSubscriber's implementation to include this maxConcurrency parameter: final int maxConcurrency; public OpFlatMap(Func1<? super T, ? extends Observable<? extends R>> mapper, int prefetch, int maxConcurrency) { this.mapper = mapper; this.prefetch = prefetch; this.maxConcurrency = maxConcurrency; } @Override public Subscriber<T> call(Subscriber<? super R> t) { FlatMapSubscriber<T, R> parent = new FlatMapSubscriber<>(t, mapper, prefetch, maxConcurrency); parent.init(); return parent; } As a contract, we will handle Integer.MAX_VALUE as an indicator for the original unbounded mode: final int maxConcurrency; public FlatMapSubscriber(Subscriber<? super R> actual, Func1<? super T, ? extends Observable<? extends R>> mapper, int prefetch, int maxConcurrency) { this.actual = actual; this.mapper = mapper; this.prefetch