Development skills of rxjs observable custom operator

Operators are RxJS Kuhe Observables The foundation block . It enables us to use some keywords （ function ） To perform complex operations . Operator is just getting the source Observable、 Take action on it and return to the new Observable Function of .

rxJS Of Pipeline operators pipe Allow us to link operators . Each operator accepts a Observable And return a Observable, This consistent behavior makes linking possible . Operator Back to Observable As input to the next operator .

Here is the simplest customization Operator Example .

import {
     interval, of } from 'rxjs';

function fancyOperator(source) {
    
  return source;
}

interval(1000)
  .pipe(fancyOperator)
  .subscribe((value) => console.log(value));

pipe The call does not actually execute Operator The logic of the body of a function , until Observable By subscribe until ？？ Is that right ？

pipeFromArray Just take out Operation The first function stored in the array .

And then call it , Input parameters this Is the source Observable.

Now we're right fancyOperator Add a few features .

import {
     interval, Observable } from 'rxjs';

function fancyOperator(source) {
    
  return Observable.create((observer) => {
    
    observer.next('Jerry');
    observer.complete();
  });
}

interval(1000)
  .pipe(fancyOperator)
  .subscribe((value) => console.log(value));

Here we have adopted the method of transplanting flowers and trees , In the custom Operator A new Observable, This Observable Data is only transmitted once , It enters the termination state .

Look at a custom in a real project Observable Example ：

import {
     fromEvent, interval, Observable } from 'rxjs';

import {
     filter } from 'rxjs/operators';

function filterKey(key) {
    
  return filter((event: KeyboardEvent) => event.key === key);
}

fromEvent(document, 'keyup')
  .pipe(filterKey('Enter'))
  .subscribe(
    (data) => console.log(data) // KeyboardEvent
  );

This customization Operator The input is Enter, The output is filter Return value of the call , A new function .filter Call to receive a predicate, Returns a new function .

to glance at filter Returns the value printed when the function runs ：

Put the original Observable, Transmit to filter Return the function and execute ：

filter The returned function pairs the original Observable Carry on the processing , Back to a new Observable：

Trigger filterOperator Constructor for ：

Put the new filterOperator The instance passes in the original Observable Of lift Method . This lift The operation is very simple , Create a new one Observable, Put the original Observable Hook up the new Observable Of source Field , At the same time Operator Assign instance to new Observable Example of operator Method .

So when you finally subscribe , The subscription is lift New after operation Observable example .

subscribe Will trigger filterOperator Of call Method ：