Single step debugging analysis of rxjs observable of operator

Look at this simplest piece of code ：

import {
     Observable, of } from 'rxjs';

const observable = of(1, 2, 3);

observable.subscribe((message) => {
    
  console.log(message);
}); 

Output ：

Input 1,2,3 Treated as an array , Trigger fromArray Function call ：

Because it doesn't exist scheduler call , So to enter subscribeToArray Branch ：

subscribeToArray Is implemented in a subscribeToArray.ts In the document , The namespace of this file is as follows ：internal/util

Be careful , No 8 Yes for loop , Clearly in of Function calls do not execute , It is of Back to Observable By subscribe Execution only .subscribeToArray The logic of , A simple for loop , Call... In turn in the loop body subscriber Of next Method , Last call complete Method .

subscribeToArray Return function , Stored in Observable Constructor's _subscribe Within the properties .

And then for this of Back to Observable example , call subscribe Method .

of Observable Example of _subscribe Method , It points to just subscribeToArray Return function ：

until Observable By subscribe, This function body can be executed . In the function body for Within the loop , Call one by one subscriber Of next Method ：

subscriber Not created by application developers , It is rxjs Internal maintenance and use of the framework .subscriber There is an attribute destination, Point to Safesubscriber, This safesubscriber Of _next attribute , It refers to the callback function maintained by the application developer .

The execution sequence of these functions is shown in the figure below ：

subscribe In addition to passing in a single callback function , And support error and complete Handle .

See the following example ：

import {
     Observable, of } from 'rxjs';

const observable = of(1, 2, 3);

observable.subscribe(
  (message) => {
    
    console.log(message);
  },
  () => {
    },
  () => {
    
    console.log('complete');
  }
);

complete Methods and next The execution logic of the method is similar to , The only difference is for Trigger after the execution of the loop body ：

thus it can be seen ,of Created Observable yes cold Observable.
If it is a periodic transmission of data Observable, We can also use unsubscribe Unsubscribe from it . Look at the code below ：

import {
     interval } from 'rxjs';

const observable = interval(1000);
const subscription = observable.subscribe((x) => console.log(x));
setTimeout(() => {
    
  subscription.unsubscribe();
}, 4500);

Output ：

The Observable At output 4 Stop transmitting the value after an integer .