I am wondering how RxJS operators such as .map() 'listen' to the incoming observable stream. Does .map() and other RxJS operators subscribe to the stream under the hood in order to get access to the observable values?
If I am understanding it correctly, each operator subscribes to the input observable, which gives it access to the stream of values then returns an observable of the transformed values to the next operator? Thanks for any clarification.
The short answer is yes. I've elaborated a bit on it in this answer
From my perspective, the magic of RxJS is achieved with linked lists.
Imagine you have something like this:
const src$ = new Observable(subscriber => subscriber.next(1))
const src2$ = src$.pipe(
map(/* ... */),
filter(/* ... */),
takeWhile(/* ... */)
)
An operator is a function which returns another function whose single argument is an Observable<T> and whose return type is an Observable<R>. Sometimes, T and R can be the same(e.g when using filter(), debounceTime()...).
Observable.pipe is the important part here. This method is what creates an observable linked list. The process is detailed in the linked answer, but, briefly, this is what you'd roughly get:
S1 // new Observable(...) - the HEAD node
|
S2 // map() - another observable, whose `.source` is `S1`
|
S3 // filter() - another observable, whose `.source` is `S2`
|
S4 // takeWhile() - another observable, whose `.source` is `S3`
When you subscribe(e.g src2$.subscribe()), a new chain will be created based on the existing one, this time it is the subscribers chain. Basically, a new subscriber will be created from .subscribe(subscriber), which will subscribe to takeWhile observable, which will create a takeWhile subscriber, which will subscribe to filter observable, which will create a filter subscriber, which will subscribe to map observable and so on. Finally, the subscriber from callback from new Observable(subscriber => subscriber.next(...)) will be, in this case, the map subscriber.
This is a good question. We first need to understand the behavior of Observables. Observables are cold by default. By cold it means that unless a code-unit subscribes to it, it will not execute its functionality Here is a quote from angular.io website
Observables are declarative—that is, you define a function for publishing values, but it is not executed until a consumer subscribes to it. The subscribed consumer then receives notifications until the function completes, or until they unsubscribe.
That is the reason we see the usage of rxjs pipe operator. pipe operator as the name suggests creates a pipe-line of cold rxjs operators that augment each previous operator. This declarative set of pipe-lined operators remain cold until / unless the final version of this declaration is subscribed.
So for example if you are making use of map rxjs operator, it would get triggered by the subscribe against the pipe-lined set of operators created. This pipe-line would get executed from left-to-right, thats first come first serve. Here is an example.
const req1 = of(1,2,3).pipe(map((x)={
return x * 2;
}));
req1.subscribe((x)=>{
console.log(x);
});
