[RxJS] Concat, Merge, ForkJoin

 rxjs    concat     merge    forkJoin

▌Introduction

---

▋concat

Concatenates multiple Observables together by sequentially emitting their values, one Observable after the other.

From reactivex.io

▋merge

Flattens multiple Observables together by blending their values into one Observable.

From reactivex.io

▋forkJoin

Runs all observable sequences in parallel and collect their last elements.

From Reactive-Extensions/RxJS

▌Samples

---

▋concat sample

observable1$=Rx.Observable.timer(50,500).map(

      x=> { return {from:'observable$1', value:x}; }).take(10);

 observable2$=Rx.Observable.timer(50,500).map(

      x=> {return {from:'observable$2', value:x}; }).take(10);

 Rx.Observable.concat(observable1$,observable2$)

    .subscribe(x=>{console.log(x)});

Result :

▋merge sample

observable1$=Rx.Observable.timer(50,500).map(

      x=> { return {from:'observable$1', value:x}; }).take(10);

 observable2$=Rx.Observable.timer(50,500).map(

      x=> {return {from:'observable$2', value:x}; }).take(10);

 Rx.Observable.merge(observable1$,observable2$)

    .subscribe(x=>{console.log(x)});

Result :

▋forkJoin sample

observable1$=Rx.Observable.timer(50,500).map(

      x=> { return {from:'observable$1', value:x}; }).take(10);

 observable2$=Rx.Observable.timer(50,500).map(

      x=> {return {from:'observable$2', value:x}; }).take(10);

 Rx.Observable.forkJoin(observable1$,observable2$)

    .subscribe(x=>{console.log(x)});

Result :

▌Live sample

We would like to display products from different api as following codes.

▋HTML

<div *ngFor="let prod of products; let i=index;">

        <main-product [prod]="prod">

        </main-product>

</div>

▋TS

//Delay the Observable sequence for 1 sec

let productsNew$ = this.prodSvc.getAll().delay(1000); 

//Delay the Observable sequence for 0.5 sec

let productsDiscount1$ = this.prodSvc.getDiscount1().delay(500); 

//The Observable sequence comes first

let productsDiscount2$ = this.prodSvc.getDiscount2();

prodsNoDiscount$.subscribe(prods =>{

      console.log("Get NO-DISCOUNT products!");

      this.products = this.products.concat(prods);

});

prodsDiscount20$.subscribe(prods =>{

      console.log("Get 20%-DISCOUNT products!");

      this.products = this.products.concat(prods);

});

prodsDiscount50$.subscribe(prods =>{

      console.log("Get 50%-DISCOUNT products!");

      this.products = this.products.concat(prods);

});

The above codes will display the products with this order.

Get 50%-DISCOUNT products!

Get 20%-DISCOUNT products!

Get NO-DISCOUNT products!



▋merge

We can use merge to replace the above codes and result in the same result.

this.products$ = Observable.merge(prodsNoDiscount$, prodsDiscount20$, prodsDiscount50$);

this.products$.subscribe(prods =>{

    this.products = this.products.concat(prods);

});

▋concat

But if we would like to display the products with certain order like this :

1.No discount

2.20% discount

3.50% discount

Then we can use concat to promise the observable sequence emits one by one.

this.products$ = Observable.concat(prodsNoDiscount$, prodsDiscount20$, prodsDiscount50$);

this.products$.subscribe(prods =>{

      this.products = this.products.concat(prods);

});

▋forkJoin

Furthermore, let’s try forkJoin which will will result the same result as concat. The difference between them in this sample is that the forkJoin will get the three kinds of products in prellel and execute the subscription with the last sequence of each observable. Thus all the products will diplayed on the webpage almost the same time.

//ForkJoin

let productsfj$ = Observable.forkJoin(prodsNoDiscount$, prodsDiscount20$, prodsDiscount50$);

productsfj$.flatMap(x=>x).subscribe(prods =>{

      this.products = this.products.concat(prods);

});

▌Reference

---

1. Combining multiple RxJs streams in Angular