operators操作符

Rxjs中有多达104个operators操作符，操作符的使用都要通过pipe方法。

avatar

用法示例


// RxJS v6+

import { interval } from 'rxjs';

import { bufferCount } from 'rxjs/operators';

// 创建每1秒发出值的 observable

const source = interval(1000);

// 在发出3个值后，将缓冲的值作为数组传递

const bufferThree = source.pipe(bufferCount(3));

// 打印值到控制台

// 输出: [0,1,2]...[3,4,5]

const subscribe = bufferThree.subscribe(val =>

  console.log('Buffered Values:', val)

);

源码

每个操作符方法接收一个参数，返回一个函数，以下以map操作符为例


//map函数接收两个参数，第一个是函数，第二个是this指针

//返回mapOperation函数，mapOperation函数参数为observable实例

//mapOperation函数执行返回一个处理过的observable实例

export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {

  return function mapOperation(source: Observable<T>): Observable<R> {

    if (typeof project !== 'function') {

      throw new TypeError('argument is not a function. Are you looking for `mapTo()`?');

    }

    return source.lift(new MapOperator(project, thisArg));

  };

}

export class MapOperator<T, R> implements Operator<T, R> {

  constructor(private project: (value: T, index: number) => R, private thisArg: any) {

  }

  call(subscriber: Subscriber<R>, source: any): any {

    return source.subscribe(new MapSubscriber(subscriber, this.project, this.thisArg));

  }

}

/**

* We need this JSDoc comment for affecting ESDoc.

* @ignore

* @extends {Ignored}

*/

class MapSubscriber<T, R> extends Subscriber<T> {

  count: number = 0;

  private thisArg: any;

  constructor(destination: Subscriber<R>,

              private project: (value: T, index: number) => R,

              thisArg: any) {

    super(destination);

    this.thisArg = thisArg || this;

  }

  // NOTE: This looks unoptimized, but it's actually purposefully NOT

  // using try/catch optimizations.

  protected _next(value: T) {

    let result: any;

    try {

      result = this.project.call(this.thisArg, value, this.count++);

    } catch (err) {

      this.destination.error(err);

      return;

    }

    this.destination.next(result);

  }

}