Blog Feed Post

RxJS Essentials. Part 5: The flatMap operator

In this article I’ll introduce an RxJS flatMap() operator. The previous articles in this series include:

1. Basic Terms
2. Operators map, filter, and reduce
3. Using Observable.create()
4. Using RxJS Subject

In some cases, you need to treat each item emitted by an observable as another observable. In other words, the outer observable emits the inner observables. Does it mean that we need to write nested subscribe() calls (one for the outer observable and another for the inner one)? No, we don’t. The flatMap() operator takes each item from the outer observable and auto-subscribes to it.

Some operators are not explained well in RxJS documentation, and we recommend you to refer to the general ReaciveX (reactive extensions) documentation for clarification. The flatMap() operator is better explained there, and it states that flatMap() is used to “transform the items emitted by an observable into observables, then flatten the emissions from those into a single observable”. This documentation includes the following marble diagram:
https://yakovfain.files.wordpress.com/2017/09/appd_flatmap.png?w=1520&h=740 1520w, https://yakovfain.files.wordpress.com/2017/09/appd_flatmap.png?w=150&h=73 150w, https://yakovfain.files.wordpress.com/2017/09/appd_flatmap.png?w=300&h=146 300w, https://yakovfain.files.wordpress.com/2017/09/appd_flatmap.png?w=768&h=374 768w, https://yakovfain.files.wordpress.com/2017/09/appd_flatmap.png?w=1024&h=499 1024w" sizes="(max-width: 760px) 100vw, 760px" />

As you see, the flatMap() operator takes an emitted item from the outer observable (the circle) and unwraps its content (the inner observable of diamonds) into the flattened output observable stream. The flatMap() operator merges the emissions of the inner observables so their items may interleave.

The following code listing has an observable that emits drinks, but this time it emits not individual drinks, but palettes. The first palette has beers and the second – soft drinks. Each palette is an observable. We want to turn these two palettes into an output observable with individual beverages.

function getDrinks() {

    let beers = Rx.Observable.from([   // 1
        {name: "Stella", country: "Belgium", price: 9.50},
        {name: "Sam Adams", country: "USA", price: 8.50},
        {name: "Bud Light", country: "USA", price: 6.50}
    ], Rx.Scheduler.async);

    let softDrinks = Rx.Observable.from([    // 2
        {name: "Coca Cola", country: "USA", price: 1.50},
        {name: "Fanta", country: "USA", price: 1.50},
        {name: "Lemonade", country: "France", price: 2.50}
    ], Rx.Scheduler.async);

    return Rx.Observable.create( observer => {
            observer.next(beers);     // 3
            observer.next(softDrinks);   // 4

// We want to "unload" each palette and print each drink info

  .flatMap(drinks => drinks)    // 5   
  .subscribe(  // 6
      drink => console.log("Subscriber got " + drink.name + ": " + drink.price ),
      error => console.err(error),
      () => console.log("The stream of drinks is over")

1. Creating an async observable from beers
2. Creating an async observable from soft drinks
3. Emitting the beers observable with next()
4. Emitting the soft drinks observable with next()
5. Unloading drinks from pallets into a merged observable
6. Subscribing to the merged observable

This script will produce the output that may look as follows (note that the drinks interleave):

Subscriber got Stella: 9.5
Subscriber got Coca Cola: 1.5
Subscriber got Sam Adams: 8.5
Subscriber got Fanta: 1.5
Subscriber got Bud Light: 6.5
Subscriber got Lemonade: 2.5
The stream of observables is over

To see it in CodePen visit this link.

Are there any other uses of the flatMap() operator besides unloading palettes of drinks? Another scenario where you’d want to use flatMap() is when you need to execute more than one HTTP request, where the result of the first request should be given to the second one. In Angular, HTTP requests return observables and without flatMap() this could be done (it a bad style) with nested subscribe() calls:

  .subscribe(customer => {
              .subscribe(response => this.order = response)

The method httpClient.get() returns an observable, and the better way to write the above code is by using the flatMap() operator, which auto-subscribes and unwraps the content of the first observable and makes another HTTP request:

          .flatMap(customer => this.httpClient.get(customer.orderURL))
          .subscribe(response => this.order = response);

Since a flatMap() is a special case of map(), you can specify a transforming function while flattening the observables into a common stream. In the above example, we transform the value customer into a function call httpClient.get().

TIP: In RxJS, flatMap() is an alias of mergeMap() so these two operators have the same functionality.

Let’s consider one more example of using flatMap(). This example will be a modified version of the traders-orders example used in the article “Using RxJS Subject“. This example is written in TypeScript and it uses two Subject instances:

* traders – this Subject keeps track of traders
* orders – this Subject is declared inside the class Trader and keeps track of each order placed by a particular trader.

You’re the manager who wants to monitor all orders placed by all traders. Without flatMap(), you’d need to subscribe to traders (the outer observable) and create a nested subscription for orders (the inner observable) that each subject has. Using flatMap() allows you to write just one subscribe() call, which will be receiving the inner observables from each trader in one stream.

import {Subject} from 'rxjs/Subject';
import 'rxjs/add/operator/mergeMap';

enum Action{
    Buy = 'BUY',
    Sell = 'SELL'

class Order{
    constructor(public orderId: number, public traderId: number, public stock: string, public shares: number, public action:Action){}

let traders = new Subject<Trader>();  // 1

class Trader {

    orders = new Subject<Order>();   // 2

    constructor(private traderId:number, public traderName:string){}

let tradersSubscriber = traders.subscribe(trader => console.log(`Trader ${trader.traderName} arrived`))

let ordersSubscriber = traders        // 3
  .flatMap(trader => trader.orders)   // 4
  .subscribe(ord =>      // 5
       console.log(`Got order from trader ${ord.traderId} to ${ord.action} ${ord.shares} shares of ${ord.stock}`));

let firstTrader = new Trader(1, 'Joe');
let secondTrader = new Trader(2, 'Mary');


let order1 = new Order(1, 1,'IBM',100,Action.Buy);
let order2 = new Order(2, 1,'AAPL',200,Action.Sell);
let order3 = new Order(3, 2,'MSFT',500,Action.Buy);

// Traders place orders
firstTrader.orders.next( order1);
firstTrader.orders.next( order2);
secondTrader.orders.next( order3);

1. Declare the Subject for traders
2. Each trader has its own Subject for orders
3. Starting with the outer observable traders
4. Extracting the inner observable from each Trader instance
5. The function subscribe() receives a stream of orders

In this version of the program, the class Trader doesn’t have a method placeOrder(). We just have the trader’s observable orders push the order to its observer by using the method next(). Remember, a Subject has both observable and observer.

The output of this program is shown next.

Trader Joe arrived
Trader Mary arrived
Got order from trader 1 to BUY 100 shares of IBM
Got order from trader 1 to SELL 200 shares of AAPL
Got order from trader 2 to BUY 500 shares of MSFT

In our example the subscriber just prints the orders on the console, but in a real world app it could invoke another function that would be placing order with the stock exchange for execution.

To see it in CodePen, follow this link. Stay tuned…

Read the original blog entry...

More Stories By Yakov Fain

Yakov Fain is a Java Champion and a co-founder of the IT consultancy Farata Systems and the product company SuranceBay. He wrote a thousand blogs (http://yakovfain.com) and several books about software development. Yakov authored and co-authored such books as "Angular 2 Development with TypeScript", "Java 24-Hour Trainer", and "Enterprise Web Development". His Twitter tag is @yfain

Latest Stories
Daniel Jones is CTO of EngineerBetter, helping enterprises deliver value faster. Previously he was an IT consultant, indie video games developer, head of web development in the finance sector, and an award-winning martial artist. Continuous Delivery makes it possible to exploit findings of cognitive psychology and neuroscience to increase the productivity and happiness of our teams.
The standardization of container runtimes and images has sparked the creation of an almost overwhelming number of new open source projects that build on and otherwise work with these specifications. Of course, there's Kubernetes, which orchestrates and manages collections of containers. It was one of the first and best-known examples of projects that make containers truly useful for production use. However, more recently, the container ecosystem has truly exploded. A service mesh like Istio addr...
As DevOps methodologies expand their reach across the enterprise, organizations face the daunting challenge of adapting related cloud strategies to ensure optimal alignment, from managing complexity to ensuring proper governance. How can culture, automation, legacy apps and even budget be reexamined to enable this ongoing shift within the modern software factory? In her Day 2 Keynote at @DevOpsSummit at 21st Cloud Expo, Aruna Ravichandran, VP, DevOps Solutions Marketing, CA Technologies, was jo...
Predicting the future has never been more challenging - not because of the lack of data but because of the flood of ungoverned and risk laden information. Microsoft states that 2.5 exabytes of data are created every day. Expectations and reliance on data are being pushed to the limits, as demands around hybrid options continue to grow.
Poor data quality and analytics drive down business value. In fact, Gartner estimated that the average financial impact of poor data quality on organizations is $9.7 million per year. But bad data is much more than a cost center. By eroding trust in information, analytics and the business decisions based on these, it is a serious impediment to digital transformation.
Business professionals no longer wonder if they'll migrate to the cloud; it's now a matter of when. The cloud environment has proved to be a major force in transitioning to an agile business model that enables quick decisions and fast implementation that solidify customer relationships. And when the cloud is combined with the power of cognitive computing, it drives innovation and transformation that achieves astounding competitive advantage.
Digital Transformation: Preparing Cloud & IoT Security for the Age of Artificial Intelligence. As automation and artificial intelligence (AI) power solution development and delivery, many businesses need to build backend cloud capabilities. Well-poised organizations, marketing smart devices with AI and BlockChain capabilities prepare to refine compliance and regulatory capabilities in 2018. Volumes of health, financial, technical and privacy data, along with tightening compliance requirements by...
As IoT continues to increase momentum, so does the associated risk. Secure Device Lifecycle Management (DLM) is ranked as one of the most important technology areas of IoT. Driving this trend is the realization that secure support for IoT devices provides companies the ability to deliver high-quality, reliable, secure offerings faster, create new revenue streams, and reduce support costs, all while building a competitive advantage in their markets. In this session, we will use customer use cases...
"NetApp is known as a data management leader but we do a lot more than just data management on-prem with the data centers of our customers. We're also big in the hybrid cloud," explained Wes Talbert, Principal Architect at NetApp, in this SYS-CON.tv interview at 21st Cloud Expo, held Oct 31 – Nov 2, 2017, at the Santa Clara Convention Center in Santa Clara, CA.
Andrew Keys is Co-Founder of ConsenSys Enterprise. He comes to ConsenSys Enterprise with capital markets, technology and entrepreneurial experience. Previously, he worked for UBS investment bank in equities analysis. Later, he was responsible for the creation and distribution of life settlement products to hedge funds and investment banks. After, he co-founded a revenue cycle management company where he learned about Bitcoin and eventually Ethereal. Andrew's role at ConsenSys Enterprise is a mul...
Evan Kirstel is an internationally recognized thought leader and social media influencer in IoT (#1 in 2017), Cloud, Data Security (2016), Health Tech (#9 in 2017), Digital Health (#6 in 2016), B2B Marketing (#5 in 2015), AI, Smart Home, Digital (2017), IIoT (#1 in 2017) and Telecom/Wireless/5G. His connections are a "Who's Who" in these technologies, He is in the top 10 most mentioned/re-tweeted by CMOs and CIOs (2016) and have been recently named 5th most influential B2B marketeer in the US. H...
The best way to leverage your Cloud Expo presence as a sponsor and exhibitor is to plan your news announcements around our events. The press covering Cloud Expo and @ThingsExpo will have access to these releases and will amplify your news announcements. More than two dozen Cloud companies either set deals at our shows or have announced their mergers and acquisitions at Cloud Expo. Product announcements during our show provide your company with the most reach through our targeted audiences.
DevOpsSummit New York 2018, colocated with CloudEXPO | DXWorldEXPO New York 2018 will be held November 11-13, 2018, in New York City. Digital Transformation (DX) is a major focus with the introduction of DXWorldEXPO within the program. Successful transformation requires a laser focus on being data-driven and on using all the tools available that enable transformation if they plan to survive over the long term. A total of 88% of Fortune 500 companies from a generation ago are now out of bus...
With 10 simultaneous tracks, keynotes, general sessions and targeted breakout classes, @CloudEXPO and DXWorldEXPO are two of the most important technology events of the year. Since its launch over eight years ago, @CloudEXPO and DXWorldEXPO have presented a rock star faculty as well as showcased hundreds of sponsors and exhibitors! In this blog post, we provide 7 tips on how, as part of our world-class faculty, you can deliver one of the most popular sessions at our events. But before reading...
DXWordEXPO New York 2018, colocated with CloudEXPO New York 2018 will be held November 11-13, 2018, in New York City and will bring together Cloud Computing, FinTech and Blockchain, Digital Transformation, Big Data, Internet of Things, DevOps, AI, Machine Learning and WebRTC to one location.