What and Why RxJS?
Reactive programming was first implemented by Microsoft engineer Eric Meijer while working in C#, LINQ. LINQ in the .NET framework provides the first kind of reactive experience to developers. Later in 2012, it was open-sourced and now used by all big companies including github, Netflix, etc.
RxJS implements reactive extension for TypeScript and JavaScript. ReactiveX is a combination of the observer pattern, iterator pattern, and functional programming. check reactive.io for more details.
Below attached image to view RxJS world!
- Observable pattern: Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically. In Angular, there are many scenarios where we have to deal with events, streams of events.
- Iterator Pattern: Provide a way to access the elements of an aggregate object sequentially without exposing its underlying representation.
- Functional Programming: Functional programming is a programming paradigm where we use a pure function, avoiding the shared state, and mutable data. Currying in JavaScript is one of the great examples of functional programming.
RxJS is an event-based and asynchronous program that uses the observable sequence. Observables can be subscribed by a method, changed by a set of methods, and can return an asynchronous set of streams which further changed by a set of methods.
You can see the word subscribe here, wait it looks like you are talking something like Promises?
Observable vs Promise:
In general, promises handle single events that fail or succeed. Though now there are many libraries which can make the promise more powerful.
Observable is a set of the stream that can handle no to many events. Observables are preferred over promises because they can be aborted, retry in case if it fails, and much more!
Example:
In the below code, we are creating one service ObservableDemoService which is implementing HttpClient. This service getting initial page info to load data. Method getinfo$ getting array stream of DemoInterface[] and in pipe handling error. In the component constructor injecting dependency of ObservableDemoService. Also creating an error subject errorMessageSubject to hold or show error in view. Now calling getinfo$ method of ObservableDemoService with pipe. Pipe takes a bunch of other methods. Pipe is very useful when we are performing some data messaging on the response, here we are initializing the data grid component.
Here we are creating promiseinAction which returns a resolve and reject promise when supplied true and false respectively. We are implementing ngOnInit method to call getPromise(true || false) method. We can see clearly here we have only two response success or failure.
Good News! Promises can be converted to Observables:
The good part is a promise can be converted to Observables when required!
RxJS operators that accept Observables can also accept a promise as well. You just need to change promise to observable. The best part is you don’t need to change all the code in the Angular app!
import { from } from 'rxjs';
const observable = from(promise)OR
import { defer, fromEvent, interval } from 'rxjs';
Observable.defer(p => Promise())Observable vs Subject
Subject is a special type of Observable. Observable is one of the base classes in RxJS and Subject is one of its derived types. Subjects implement observer design pattern which is required to set some value i.e. this.Subject.next(message). Observable always need a subscription and implement the observable pattern. Observable is always unidirectional meaning it flows from source to subscriber. Subject is bidirectional, information flow from the source to subscriber, and vice versa.
Types of Subject:
- BehaviourSubject: A BehaviourSubject stores the latest value and immediately sends it to all subscribers. i.e. I have created a message service that can be consumed from two Angular components and its bidirectional.
Component1:
sendMessage(message: any): void {this.messageService.changeMessage(message);}
Component2:
let getMessage;this.messageService.currentMessage.subscribe(message => (getMessage = message));
2. ReplaySubject: A replay subject stores n number of values and immediately send it to the subscriber. i.e. in the above data service if we change messageSource as ReplaySubject and pass 3 then it will cache the last three values on subscription.
private messageSource = new ReplaySubject<string>(3);3. AsyncSubject: Async subject waits until it completely and then sends the final value to the subscriber. The keyword here is the final value. If we add two values in the below example “Hello 1” and “Hello Again”, When we do asyncsub.complete() then only last value will be printed. The last value is the final value in our case.
import { AsyncSubject } from ‘rxjs’;const asyncsub = new AsyncSubject();asyncsub.subscribe(subs => console.log(subs));asyncsub.next("Hello 1");asyncsub.next("Hello Again");asyncsub.complete();//Logged only Hello Again NOT Hello 1
4. ConnectableObservable: A wrapper that transforms observable behave like a Subject. It provides a .connect() method to transform. Very less likely you will be using it in Angular application.
observable.connect();
Managing Stream Greater Power !!
RxJS operators provide managing streams, combining the stream into a single stream. It also helps in calling APIs (set of observables) sequential and parallel. But the main problem is there are so many operators. Combining operators is always a challenge. RxJs provides a set of the wizard on their website for combining these operators. If the wrong operator is combined it will lead to race conditions. Also, all the subscriptions need to unsubscribe when components are out of scope, failing this will led to memory leaks. Memory leaks are really really scary especially when the application is running in production environments.
Combining Observables into one Stream:
- mergeMap: Creates new observables for any given source. All previous streams/observable keep alive. There is no order in the returned observables, moreover, the order is not preserved. The best use case for mergeMap is when combining a click event with API calls.
- concatMap: Similar to mergeMap but the order of observables is well preserved. Preserve the order and emits all observable value, works synchronously. Execute slowly because it works synchronously, waits for first observable to complete then only start new observables stream. The best use case is when you are calling an API which gives you id and that id is used in another API.
- switchMap: Immediately creates new observables and completes the old observables. The best use case for switchMap is search auto-complete. Whenever the user starts typing a new keyword for search, a new observable is created and the old one is completed. Check combineLatest example below with mergeMap and switchMap.
- flatMap: Immediately creates observables and previous observables are kept alive. fatmap is an alias of mergemap, mergeMap accepts an optional parameter concurrency, which defines how many Observables can be subscribed at the same time.
- exhaustMap: Creates observable and waits for it until it complete. All other observable is ignored while waiting for the observable to complete. Best use case is to use login in the Angular app. Once the user clicks on login then wait until authentication is done! exhaustMap is just the opposite of switchMap. switchMap immediately creates a new observable and completes the old ones however exhaustMap first complete the initial observable and ignores the new ones.
Joining Observables into Array:
- forkJoin: Calls all observable parallel. Returns all observables as an array once all call is completed. The best use case is when you want to call APIs that are not depends on each other. Multiple uploads are one of the scenarios where we can use forkJoin. Also, we can combine insert and update APIs in forkJoin based on conditions.
ngOnInit() {const promiseList = [];promiseList.push(this.myService.getData1());promiseList.push(this.myService.getData2());if (promiseList.length) {forkJoin(promiseList).subscribe(resultList=>{//Initialize component data here
this.dataModel.data1= resultList[0]})}}
- combineLatest: Begins when all observables fired at least once. Afterwards it fires when any of the event changes.
The best use case in an Angular application is combining a data table and dropdown events action. Dropdown represents the product with category and category ids and once selected we have to show product details based on category id. Now check the below code:
products$ =combineLatest([this.productService.product$,this.action$]).pipe(map(([products,category])=>products.filter(x=>x.categoryId===category)));
Once merged with combined with the latest whenever the user selects a product category dropdown data table data automatically filtered. No need to write any event in Angular component. This is sometimes called a reactive style of development.
Now we can use mergeMap while filtering the selectedCategory. If the user is clicks on the drop-down again and the previous request is not completed, then switchMap will cancel the previous request and start a new Observables.
selectedCategoryCliks$ = this.selectedCategory$ .pipe( filter(selectedCategory => Boolean(selectedCategory)), switchMap(selectedCategory => from(selectedCategory.Id) .pipe( mergeMap(Id => this.http.get<Interface>(`${environment.getURL}`)), toArray(), tap(x => console.log(‘’) ) ));
With great power comes great responsibility! Memory Leaks are real!
I know you already heard this phase. But in the RxJS world, it’s very important to always unsubscribe to open observables and subjects. In Angular, implement ngOnDestroy and unsubscribe.
ngOnDestroy(): void { this.subject.unsubscribe();}
However many times it is difficult to remember which teardown method to unsubscribe.
How about if the subscription is automatically unsubscribed rather than doing it manually?
It is possible in frameworks like Angular if we are using async pipes. Async pipes automatically unsubscribe open observables and subject once a component is out of scope. Now we already created products$ in the above example which holds the product list coming from getting service. Using product$ | async pipe we can show the product details. Below is the view code snippet.
<ng-container *ngIf=”products$ | async as productList”> <tr *ngFor=”let prod of productList; let i = index” class=”tr-clickable” (click)=”clickonRow(prod)”> <td *ngIf=”prod.Name”><span style=”display: none;”>{{ prod.Name }}</span> </td> </tr></ng-container>
No comments:
Post a Comment