RxJS | Create Custom Observables from event sources directly
RxJs simplifies working with event streams. In Angular, we get notified of almost all events and changes by subscribing to RxJs Observable(s)
Ex (ActvatedRoute#params , HttpClient#get).
We seldom create our own Observable(s)
from an actual event source. Unless, you consider emitting known values, with of and from as we usually do in our tests.
import { Observable, of, from } from 'rxjs';Observable<String> one = of('1');
Observable<String> oneTwoThree = from(['1','2', '3']);
RxJs provides us handy utility function from(...) : Observable<T>
to create Observable(s)
from well known event sources, (Ex. a native dom-event).
Here is an example of fromEvent(input[click]):Observable
, implementing a text-input search-box, which can notify us of changing text.
import {AfterViewInit, Component, ElementRef, EventEmitter, OnDestroy, OnInit, Output, ViewChild} from '@angular/core';
import {fromEvent, Observable, Subscription} from 'rxjs';
import {debounceTime, distinctUntilChanged, map, startWith, tap} from 'rxjs/operators';@Component({
selector: 'search-box',
template: `
<div>
<input placeholder="search" #searchInput autocomplete="off"/>
</div>
`
})
export class SearchBoxComponent implements AfterViewInit, OnDestroy {@Output('onSearch')
onSearch = new EventEmitter<string>();@ViewChild('searchInput')
input: ElementRef;private subscription: Subscription;ngAfterViewInit(): void {
const terms$ = fromEvent<any>(this.input.nativeElement, 'keyup')
.pipe(
map(event => event.target.value),
startWith(''),
debounceTime(400),
distinctUntilChanged()
);
this.subscription = term$
.subscribe(
criterion => {
this.onSearch.emit(criterion);
}
);}ngOnDestroy() {
this.subscription.unsubscribe();
}}
Custom Observable(s)
Sometimes source of your event(s) is not well known, and likely RxJs wouldn’t have any stock functions to create Observable(s)
of it.
RxJs provides a mechanism to create our own Observable(s).
Observable.create(function(observer) {
// create or listen to an event-source (ex promises)
// decide when to call observer.(next|error|complete)
})
For example, Let us try creating our own Observable
the works like Angular’s Http.get
using the browser’s native fetch
-api
import {Observable} from 'rxjs';
export function createHttp$(url:string) {
return Observable.create(observer => {
const controller = new AbortController();
const signal = controller.signal;
fetch(url, {signal})
.then(response => {
if (response.ok) {
return response.json();
}
else {
observer.error(`Failed HTTP : response.status`);
}
})
.then(body => {
observer.next(body);
observer.complete();
})
.catch(err => {
observer.error(err);
//observable which immediately errors out
});
return () => controller.abort()
// this return function? executed on unsubscribe
});
}
This is how we can use it
const http$ = createHttp$('/some/url');
http$.subscribe({
next: (value: any) => console.log(`next ${value}`),
complete: () => console.log(`complete`)};
);
Everything in the above code-snippet is straightforward besides the function returned by Observable.create(…)
which is the one that’s called when you unsubscribe to the observable created created by createHttp$(...)
.
Browser’s fetch(…)
API gives us way to cancel ongoing requests by sending an abort signal. Therefore we send the abort signal when the subscriber unsubscribes from on Observable before it completes. This will cancel long running http-get-request to which no one now is subscribing.
const url = '/some/entity';
const http$ = this.activatedRoute.params.pipe(
switchMap( id => createHttp$(`${url}/${id}`))
).subscribe({
next: (value: any) => console.log(`next ${value}`),
complete: () => console.log(`complete`)};
);
In the example above, we listen to changing URL parameters emitted from Router’s
from activatedRoute.params
and issue corresponding http requests with newly emitted param value {id}
using Observables(s)
created by callingcreateHttp$(`${url}/${id}`))
However since we are using switchMap(…)
to emit new createHttp$(…)
Observales
, the previously emitted Observable
is first unsubscribed (cancelled) before a new one is created and emitted by switchMap(…)
.
Since Observable(s)
created from createHttp$(...)
implements and returns function for cancellation, if http request inside was still on going in the Observable
being unsubscribed, it will get aborted before new createHttp$(...)
Observable
is created and emitted by switchMap
with in which a new http request gets issued.