[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

[RxJS] switch and switchMap

 rxjs    switch     map    switchMap

▌Introduction

---

▋switch

Transforms an observable sequence of observable sequences into an observable sequence producing values only from the most recent observable sequence.

From reactivex.io

▋switchMap

= map + switch

▌switch sample

---

Let’s create a sample with this requirement:

Show a user’s last mouse position.

observable$ = Rx.Observable.fromEvent(document.body, 'mousemove')

                  .map(m=> "X:"+ m.x + ", Y:"+m.y);

 subscription$ = observable$.subscribe(position=>console.log(position));

▋Demo

Now let’s specify the requirement as …

Show a user’s last mouse position when he/she stop moving it.

Okay, this time we don’t want to show every position with the Observable sequence but only the LAST one.

We will use map, delay and switch to finish this job.

observable$ = Rx.Observable.fromEvent(document.body, 'mousemove')

                  .map(m=> "X:"+ m.x + ", Y:"+m.y);

 subscription$ = observable$

                 .map(position=> Rx.Observable.of(position).delay(2000))

                 .switch()

                 .subscribe(position=>console.log(position));

The following sequence picture shows how the above codes works.

▋Demo

▌switchMap sample

---

switch = map + switch

So we can update the above sample codes to

observable$ = Rx.Observable.fromEvent(document.body, 'mousemove')

                  .map(m=> "X:"+ m.x + ", Y:"+m.y);

 subscription$ = observable$

                 .switchMap(position=> Rx.Observable.of(position).delay(2000))

                 .subscribe(position=>console.log(position));

▋Sample

We will make another practice on the searching product (again!).

We would like to search the products when the user input some keywords.

ngAfterViewInit() {

    const elKeyword = document.getElementById('keyword');

    const keyups$ = Observable.fromEvent(elKeyword, 'keyup').map((e: any) => e.target.value);

    keyups$.subscribe((keyword: string) => {

      console.log("Searching : "+ keyword);

    })

  }

The above codes will make a disaster for searching every words we key in…

Yes! We could use switchMap and delay for just taking the last keywords and search it.

ngAfterViewInit() {

    const elKeyword = document.getElementById('keyword');

    const keyups$ = Observable.fromEvent(elKeyword, 'keyup').map((e: any) => e.target.value);

    keyups$.switchMap(this.delaySearch).subscribe((keyword: string) => {

      console.log("Searching : "+ keyword);

      this.products$ = this.prodSvc.search(keyword);

    })

  }

private delaySearch(keyword: string): Observable<string> {

    //return observale

    return Observable.of(keyword).delay(1000);

    //Or return promise

    // return new Promise((resolve, reject) => {

    //   setTimeout(() => {

    //     resolve(keyword);

    //   }, 1000);

    // });

  }

▋Demo

▌Reference

---

1.RxJS : map

2.RxJS : Delay

3.RxJS : switch