操作符文档

API

create

const { Observable } = require('rxjs');

// 创建 Observables

var observable = Observable.create(observer => {

    var id = setInterval(() => {

        subscriber.next(123);

        subscriber.error('err: ...');

        subscriber.complete('done.');

    }, 1000);

    // 提供取消和清理 interval 资源的方法

    return function unsubscribe() {

        clearInterval(id);

    };

});

// 订阅 Observables

var subscription = observable.subscribe(

  v => console.log(v),

  e => console.log(e),

  done => console.log(done),

);

// .add() 合并多个订阅，以便同时取消订阅

// subscription.add(childSubscription);

// 撤销 add的合并

// subscription.remove(childSubscription);

// 3s后取消订阅

setTimeout(() => {

    subscription.unsubscribe();

}, 3000)

of(...items) 和 from([...items])

按顺序发出每个元素, from 可以代替 fromPromise

let { of, from } = require("rxjs");

of(1, 2, 3).subscribe(v => l(v)); // 1,2,3

from([1, 2, 3]).subscribe(v => l(v));// 1,2,3

bindCallback

把普通的回调函数转化为 Observable

let { bindCallback } = require("rxjs");

function test(cb) { cb(1); }

let o$ = bindCallback(test)();

o$.subscribe(v => l(v)); // 1

bindNodeCallback

把标准的node回调函数转化为 Observable

let { bindNodeCallback } = require("rxjs");

let fs = require("fs");

// fs.readFile("./test.jss", "utf8", (e, d) => (!!e ? l(e) : l(d)));

let o$ = bindNodeCallback(fs.readFile)("./test.jss", "utf8");

o$.subscribe(v => l(v), e => l(e));

combineLatest

组合多个Observable以创建一个Observable, 其值是根据每个输入Observable的最新值计算

let { combineLatest, of, timer } = require("rxjs");

let o1$ = of(233);

let o2$ = of(12);

let o3$ = timer(0, 1000);

let o$ = combineLatest(o1$, o2$, o3$);

// let o$ = combineLatest([o1$, o2$, o3$]); // 和上面一样

o$.subscribe(v => l(v));

// [ 233, 12, 0 ]

// [ 233, 12, 1 ]

// [ 233, 12, 2 ]

// ...

empty

仅仅发出 complete 通知，其他什么也不做

let { empty } = require("rxjs");

const l = console.log;

empty().subscribe(l, l, () => l("done")); // done

forkJoin

连接传递的Observables发出的最后一个值。

let { forkJoin, of } = require("rxjs");

forkJoin(of(1), of(2, 3)).subscribe(l); // [ 1, 3 ]

merge

把多个流的合并到一个 流, 按顺序执行，异步流会被延后

let { merge, of, from } = require("rxjs");

merge(of(1), of(4), from(["a", "b"]), 1).subscribe(l); // 1 4 a b

concat

它顺序地从给定的Observable中发出所有值，必须等上一个完成才执行下一个

let { concat, interval, of } = require("rxjs");

let { take } = require("rxjs/operators");

const l = console.log;

let o1$ = interval(1000).pipe(take(4)); // take 接收源 最初的N个值

let o2$ = of(233);

let o$ = concat(o1$, o2$);

o$.subscribe(v => l(v)); // 0,1,2,3,233

never

创建一个不向观察者发出任何项的 Observable

let { never } = require("rxjs");

never().subscribe(l, l, () => l("done")); //

range

range(start: number = 0, count: number = 0)

let { range } = require("rxjs");

// 从2开始 发出后续 11位数字

range(2, 11).subscribe(l); // 2..12

throwError

仅仅发出 error 通知

let { throwError } = require("rxjs");

throwError("err message").subscribe(l, l, () => l("done")); // err message

timer

let { timer } = require("rxjs");

// 延迟3秒发送第一个值，然后马上结束

timer(3000).subscribe(

  v=> console.log(v))

//每隔3s发送一个值，第一个值延迟5s

timer(5000, 3000).subscribe(

  v=> console.log(v))

interval

在指定的时间间隔内发出序列号 0+

let { interval } = require("rxjs");

let { take } = require("rxjs/operators");

interval(200)

  .pipe(take(3))

  .subscribe(l); // 0,1,2

defer

懒惰地创建Observable

let { defer, of } = require("rxjs");

let o$ = defer(() => Math.random() > 0.5 ? of(1) : of(2) );

o$.subscribe(v => l(v));

zip

其值根据其每个输入Observable的值按顺序计算

let { zip, of } = require("rxjs");

let { map } = require("rxjs/operators");

let age$ = of(27, 25, 29);

let name$ = of("Foo", "Bar", "Beer");

let isDev$ = of(true, true, false);

zip(age$, name$, isDev$)

  .pipe(map(([age, name, isdev]) => ({ age, name, isdev })))

  .subscribe(l);

// res

{ age: 27, name: 'Foo', isdev: true }

{ age: 25, name: 'Bar', isdev: true }

{ age: 29, name: 'Beer', isdev: false }

iif

iif(test, then, else)

let { iif, of } = require("rxjs");

iif(() => Math.random() > 0.5, of("then"), of("else")).subscribe(l); // then | else

fromEvent

let { fromEvent } = require("rxjs");

const clicks$ = fromEvent(document, 'click');

clicks$.subscribe(x => console.log(x));

generate

仔细看有点像 for循环

let { generate } = require("rxjs");

generate(0, x => x < 3, x => x + 1).subscribe(

  v => l(v),

  err => {},

  () => l("done"),

);

// 0, 1, 2, done

identity

没有做处理，返回值本身

let { identity } = require("rxjs");

let x = identity(233);

l(x); // 233

isObservable

let { isObservable } = require("rxjs");

l(isObservable(2)); // false

onErrorResumeNext

将无错误地移动到下一个源

let { onErrorResumeNext, of } = require("rxjs");

let { map } = require("rxjs/operators");

onErrorResumeNext(

  of(1, 2, 0, 1).pipe(

    map(el => {

      if (el === 0) throw Error("000");

      return el + el;

    }),

  ),

  of("2"),

).subscribe(l, err => console.error(err), () => l("done")); // 2, 4, 2, done

pairs

将对象转换为Observable

let { pairs, of } = require("rxjs");

pairs({

  name: "ajanuw",

  age: 14,

}).subscribe(l); // [ 'name', 'ajanuw' ] [ 'age', 14 ]

race

返回，第一个Observable的反映的输出，然后结束

let { race, of, interval } = require("rxjs");

race(interval(1000), of(2, 22), of(3)).subscribe(l, l, () => l("done")); // 2， 22， done