Angular - Using RxJS Operators mergeMap and concatMap

The Angular MergeMap and ConcatMap are used to map each value from the source observable into an inner observable. It internally subscribes to the source observable, and then starts emitting the values from it, in palce of the original value. A new inner observable will be created for every value it receives from the Source. It merges the values from all of its inner observables and emits the values back into the stream.

Difference between MergeMap and ConcatMap is that ConcatMap maintains the order of its inner observables, while MergeMap can process the source observables in any order depending on the execution time-period of each observable.

ConcatMap operator

ConcatMap processes the source observables in a serialized fashion waiting for each one to complete before moving to the next.

Lets see any example:

////on the top of the file, import the operator
//import { concatMap } from 'rxjs/operators';

//an observable of numbers (milliseconds), we use as its values to cause the delay 
const source = of(2000, 1000);

// map value from source into inner observable, once its completes, then it will move to next value in the source observable
const newSource = source.pipe(
  concatMap(val => of(`Delayed by: ${val} ms`).pipe(delay(val))) //creates a new observable, the following section is actually subscribe to this observable
);

//subscribe to the new observable (internally created by concatMap)
const subscribe = newSource.subscribe(val =>
  console.log(`With concatMap: ${val}`)
);

This code is performing the following actions:

• A source observable is defined with two values 2000 and 1000 representing milliseconds.
• concatMap is used to receive values from source and emit its own values from the new observable. Before emitting the value, the inner observable is calling the delay function to simulate delay in execution. As per the values provided in the source observable, first value cause delay for 1 second and second value will cause a delay for 2 seconds.
• The new observable is assigned to the variable newSource.
• In the end, we subscribe to the newSource observable, and write the output to the console.

Here is the sample output from above code:

With concatMap: Delayed by: 2000 ms 
With concatMap: Delayed by: 1000 ms

From this output, its clear that the concatMap will keep the original order of values emitted from the source. The values are being displayed in the same order as we supplied in the source observable. Even the second value has shorter delay of 1 second, but it will wait for the first value (with longer delay of 2 seconds) to complete before moving to next value in the source observable.

The concatMap assures the original sequence of values. If we have multiple inner Observables, the values will be processed in sequential order. Next value will only be processed after the previous one is completed.

MergeMap operator

MergeMap is similar to the contactMap with one difference that it processes the source observables without any assurance of the order of provided values.

Lets see any example, we use the same source observable (used in above exmaple):

////on the top of the file, import the operator
//import { mergeMap } from 'rxjs/operators';

//an observable of numbers (milliseconds), we use as its values to cause the delay 
const source = of(2000, 1000);

// map value from source into inner observable, it will move to next value (whenever available) in the source observable, will not wait for the previous one to complete
const newSource = source.pipe(
  mergeMap(val => of(`Delayed by: ${val} ms`).pipe(delay(val))) //creates a new observable, the following section is actually subscribe to this observable
);

//subscribe to the new observable (internally created by mergeMap)
const subscribe = newSource.subscribe(val =>
  console.log(`With mergeMap: ${val}`)
);

This code is performing the following actions:

• A source observable is defined with two values 2000 and 1000 representing milliseconds.
• mergeMap is used to receive values from source and emit its own values from the new observable. Before emitting the value, the inner observable is calling the delay function to simulate delay in execution. As per the values provided in the source observable, first value cause delay for 1 second and second value will cause a delay for 2 seconds.
• The new observable is assigned to the variable newSource.
• In the end, we subscribe to the newSource observable, and write the output to the console.

Here is the sample output from above code:

With mergeMap: Delayed by: 1000 ms 
With mergeMap: Delayed by: 2000 ms

From this output, we know that the mergeMap do not keep the original order of values emitted from the source. The values can be displayed in the order of their execution time-period. The sooner the value is processed, it will be emitted to the subscription. Since the second value has shorter delay of 1 second, it does not wait for the first value (with greater delay of 2 seconds) to complete, and hence it gets processed before the first.

The mergeMap does not assure the original sequence of values. If we have multiple inner Observables, the values may be overlapped over time, because values will be emitted in parallel.

References:

• RxJS - concatMap operator
• RxJS - mergeMap operator

Related Post(s):

• Angular - Using Takeuntil RxJS Operator
• Angular - Using Takeuntil RxJS Operator in Base Class
• Angular - Using RxJS Operators take and skip

Angular - Using RxJS Operators take and skip

take and skip operators are used to limit the number of values emitted from the source observable.

take operator

take operator returns the observable which will limit the number of values emitted and receive the first n number of values. It will take a count argument representing the max number of values expecting to be received. Usually it is being used by passing 1 as count argument, to take only the first value emitted from an observable. After receiving the n number of values, it will complete the observable, so any more values emitted after the first one will be ignored (or not received).

Lets see this example:

const sourceObservable = of(1, 2, 3, 4, 5);
const wrapperObserable = sourceObservable.pipe(take(1));
const subscribe = wrapperObserable.subscribe(val => console.log('Received Value: ' + val));

The output will be:

Received Value: 1

Here, we created a source observable which will emit values 1,2,3,4,5. Then we used the take operator with count argument as 1, and the wrapper observable will be able to emit only 1 value, hence the subscription will receive 1 value.

Lets change the count argument to 3:

const wrapperObserable = sourceObservable.pipe(take(1));
const subscribe = wrapperObserable.subscribe(val => console.log('Received Value: ' + val));

This time, the output will be:

Received Value: 1
Received Value: 2
Received Value: 3

Note that, we are not making any changes to the source observable, but we changed the count argument to the take operator to receive the desired number of values.

skip operator

skip operator also returns the observable which will limit the number of values emitted, but it works in opposite to the take operator. It will ignore the first n number of vaues and receive all of the remaining values. It will take a count argument representing the max number of values expecting to be skipped.

Lets see this example:

const sourceObservable = of(1, 2, 3, 4, 5);
const wrapperObserable = sourceObservable.pipe(skip(1));
const subscribe = wrapperObserable.subscribe(val => console.log('Received Value: ' + val));

The output will be:

Received Value: 2
Received Value: 3
Received Value: 4
Received Value: 5

Here, we created a source observable which will emit values 1,2,3,4,5. Then we used the skip operator with count argument as 1, and the wrapper observable will be able to skip 1 value, and the subscription will receive remaining all values 2,3,4,5.

Lets change the count argument to 3:

const wrapperObserable = sourceObservable.pipe(skip(1));
const subscribe = wrapperObserable.subscribe(val => console.log('Received Value: ' + val));

This time, the output will be:

Received Value: 4
Received Value: 5

References:

• RxJS - take operator
• RxJS - skip operator

Related Post(s):

• Angular - Using Takeuntil RxJS Operator
• Angular - Using Takeuntil RxJS Operator in Base Class

Angular - Using Takeuntil RxJS Operator in Base Class

In the last post , we have seen how to use takeuntil operator to automatically unsubscribe from an observable. takeuntil operator makes it easier to manage the control to unsubscribe from multiple observables.

In that exmaple we have implemented the takeuntil operator in a single component. If you need to implement the same technique in multiple components then you have to repeat the same logic in every component.

In this post we will see how we can implement takeUntil operator in a base class, so that we dont have to repeat the similar code in multiple components.

Here is the code for base class:

import { Subject } from 'rxjs';
import { Component, OnDestroy } from '@angular/core';


@Component({
    template: ''
})
export abstract class BaseComponent implements OnDestroy {

protected componentDestroyed$ = new Subject();

    constructor() { }

    ngOnDestroy() {
        this.componentDestroyed$.next();
        this.componentDestroyed$.complete();
    }
}

And the ChildComponent inheriting the BaseComponent class defined above.

import { Component, OnInit, OnDestroy } from '@angular/core';
import { takeUntil } from 'rxjs/operators';
import { Service1 } from 'Service1';
import { Service2 } from 'Service2';
import { BaseComponent } from 'src/app/models/base-component.model';

@Component({ ... })
export class ChildComponent extends BaseComponent implements OnInit  {

  constructor(private myservice1: Service1, private myservice2: Service2) {}

  ngOnInit() {
    
    this.myservice1.getData()
    .pipe(takeUntil(this.componentDestroyed$)) //componentDestroyed$ is defined in BaseComponent
    .subscribe(({data}) => {
      console.log(data);
    });
	
    this.myservice2.getData()
    .pipe(takeUntil(this.componentDestroyed$))
    .subscribe(({data}) => {
      console.log(data);
    });	
  }
 
 }

Note that, we don't need to implment OnDestroy (ngOnDestroy handler) in ChildComponent to call the next() and complete() methods for subject componentDestroyed$, because we have already defined this in the BaseComponent.

References:

• Learn RxJS - takeUntil

Related Post(s):

• Angular - Using Takeuntil RxJS Operator
• Observables vs Subjects vs Behavior Subjects

Angular - Using Takeuntil RxJS Operator

In the last post , we have seen different types of observable. Observable is basically a container which produces asynchronous stream of data, and emit values over time. We have to subscribe to an observable in order to consume or receive data. But you have to be careful with observables as it may leads to memory leaks and affect the application performance. To avoid this issue, one approach is to keep the reference at the time of subscription and unsubscribe from the observable by using the same reference when you no longer need to receive observable's data stream.

Lets see the example code:

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs';
import { Service1 } from 'Service1';
import { Service2 } from 'Service2';

@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {
  mySubscription1: Subscription;
  mySubscription2: Subscription;

  constructor(private myservice1: Service1, private myservice2: Service2) {}

  ngOnInit() {
    
    this.mySubscription1 = this.myservice1.getData()
    .subscribe(({data}) => {
      console.log(data);
    });
	
    this.mySubscription2 = this.myservice2.getData()
    .subscribe(({data}) => {
      console.log(data);
    });
	
  }
 
  ngOnDestroy() {
    this.mySubscription1.unsubscribe();
    this.mySubscription2.unsubscribe();
  }
 
 }

In this code snippet, we have two services to consume Service1 and Service2. In ngOnInit handler, we maintained the subscription references for both the services in two different variables mySubscription1 and mySubscription2. Then in the ngOnDestroy handler, we are using the same reference variables to unsubscribe from the observable.

The above code works perfectly, but it would be cumbersome to maintain in long term when you have more number of subscriptions which makes you to keep the references for each subscription and then unsubscribe from each one in ngOnDestroy handler.

A better approach is to use takeUntil operator from RxJS library, it is used to automatically unsubscribe from an observable. takeUntil refelects the source Observable. It also monitors a second Observable (the notifier) that you provide. If the notifier emits a value, the output Observable stops reflecting the source Observable and completes itself.

Here is the same exmaple, this time unsubscribe using takeUntil operator.

import { Component, OnInit, OnDestroy } from '@angular/core';

import { Subject, interval } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
import { Service1 } from 'Service1';
import { Service2 } from 'Service2';

@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {
  destroy$: Subject = new Subject();

  constructor(private myservice1: Service1, private myservice2: Service2) {}

  ngOnInit() {

    this.myservice1.getData()
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
	
    this.myservice2.getData()
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
  }

  ngOnDestroy() {
    this.destroy$.next(true);
    this.destroy$.unsubscribe();
    
    ////some people prefer to call complete() on destroy$ here, instead of unsubscribe()
    //this.destroy$.complete();
  }
} 

This code snippet also behaves the same as before, but its easier to manage when you have more number of subscriptions. In ngOnDestroy handler, we have called the unsubscribe() method, some people may prefer to call complete() method instead. But in the end it does not make much difference in this context, the purpose here is just to stop receiving more values from this subject.

References:

• Learn RxJS - takeUntil
• Observables vs Subjects vs Behavior Subjects

Angular - Validate autocomplete against available options

I used the angular-ng-autocomplete for dropdown lists and its being quite useful and works extremely good with filter.

The issue I faced with validation and it does not behave as per the expectation.

When a user searches for an option by entering the keyword text (but doesn't pick any of the available options), then the required validator failed here. If the input box is empty then the requried validator is working fine on its way. But when the input box has some text, then the requried validator will not trigger. The ng-autocomplete do not have the chance to raise the selected event, and so do not properly set the binded control's value. Since the input box has some value, the form will pass the validation, and when submitted(with null value) it will take the undefined value for the control to the server API.

The html for ng-autocomplete is:

<div class="ng-autocomplete">
	<ng-autocomplete
	  [data]="citiesList"
	  [searchKeyword]="cityName"
	  formControlName="CityId"
	  (selected)="citySelected($event)"
	  (inputChanged)="onChangeCitySearch($event)"
	  [itemTemplate]="itemTemplate"
	  [notFoundTemplate]="notFoundTemplate"
	>
	</ng-autocomplete>

	<ng-template #itemTemplate let-item>
	  <a [innerHTML]="item.Name"></a>
	</ng-template>

	<ng-template #notFoundTemplate let-notFound>
	  <div [innerHTML]="notFound"></div>
	</ng-template>
</div>

In .ts file, I am filling the citiesList from the API:

onChangeCitySearch(search: string) {
   
    //if user has entered at-least 2 characters, then call the api for search
    if (search && search.trim().length >= 2) {
      this.commonDataService
        .getCities(search)
        .subscribe((res) => {
          this.citiesList = res.Data;
        });
    }
  }

I do not want to permit the user to post the form unless one of the suggested options is selected from the list. I fixed the issue by defining a custom validator.

We could have two possible scenarios with ng-autocomplete when validating against a list of options:

• Array of strings - Available options are defined as an array of strings.
• Array of objects - Available options are as (an object property i.e. id, name etc, defined on) an array of Objects.

Bind with Array of strings

To validate autocomplete against an array of string options, we can pass the array of options to the the validator, and check if the control's value is exists in the array.

function autocompleteStringValidator(validOptions: Array<string>): ValidatorFn {
  return (control: AbstractControl): { [key: string]: boolean } | null => {
    if (validOptions.indexOf(control.value) !== -1) {
      // null means we dont have to show any error, a valid option is selected
      return null;
    }
	
    //return non-null object, which leads to show the error because the value is invalid
    return { match: false };
  }
}

This is how we can add the validator to the FormControl along with other built-in validators.

public cityControl = new FormControl('', 
    { validators: [Validators.required, autocompleteStringValidator(this.citiesList)] })

Bind with Array of Objects

We can validate the controls value when its binds to an array of objects by using the same technique as above. But I will use a slightly different version, instead of checking the index of input value in the array, here I am using filter method to find the matching item. If it founds any matching record, then the user has properly selected an option from the given list.

function autocompleteObjectValidator(myArray: any[]): ValidatorFn {
    return (control: AbstractControl): { [key: string]: boolean } | null => {
    let selectboxValue = control.value;
    let matchingItem = myArray.filter((x) => x === selectboxValue);

    if (matchingItem.length > 0) {
        // null means we dont have to show any error, a valid option is selected
        return null;
    } else {
        //return non-null object, which leads to show the error because the value is invalid
        return { match: false };
    }
    };
}

The good thing about this technique is that, you can also check for any particular property of the object in the if condition. Lets suppose, if the object has a property Id, we can check if the value of Id is matched on both objects.

let matchingItem = myArray.filter((x) => x.Id === selectboxValue.Id);

Another simpler technique can be applied by checking the type of control.value. For a valid option being selected from the list of objects, its type will be object, and in case the user types the text manully, than the type of control.value will be a simple string. So we can check, if the type is string, then it shows the fact that user has not selected any of the available options from objects list.

function autocompleteObjectValidator(): ValidatorFn {
  return (control: AbstractControl): { [key: string]: boolean } | null => {
    if (typeof control.value === 'string') {
        //return non-null object, which leads to show the error because the value is invalid
        return { match: false };
    }
	
    // null means we dont have to show any error, a valid option is selected
    return null;
  }
}

References:

• Angular Autocomplete