Rxjs expand的用法分析

Rxjs的expand()函数声明：

public expand(project: function(value: T, index: number), concurrent: number, scheduler: Scheduler): Observable

expand()会递归调用project函数，project函数把源值映射为一个Observable，每次递归调用都是把前一次调用输出的Observable的源作为输入。最后把所有的Observable合并为一个Observable作为结果输出。

它接收三个参数：

• project：映射函数，以前一次调用输出额Observable作为输入，返回一个新的Observable。
• concurrent：最大并发次数，默认值为Number.POSITIVE_INFINITY，即无限次递归调用
• scheduler：默认为null，表示立即执行

单个源值

示例：

const source = Rx.Observable.of(1);
const powerTow = source
 // 递归调用提供的函数
 .expand(val => {
  console.log(`输入: ${val}`);
  return Rx.Observable.of(2*val);
 })
 .take(5);
const subscribe = powerTow.subscribe(val => console.log(`输出: ${val}`));

示例输出源值1，每次对源值乘以2。take(5)限定了只输出前5个Observable，包括初始的源值1。

结果为：

"输出: 1"
"输入: 1"
"输出: 2"
"输入: 2"
"输出: 4"
"输入: 4"
"输出: 8"
"输入: 8"
"输出: 16"
"输入: 16"

多源值

单个源值是比较容易理解，递归调用project函数是顺序的。当源值产生是异步且是多个的，每一个源值都会单独递归调用project映射函数。最后把所有产生的Observable合并为一个Observable作为结果。

示例：

var clicks = Rx.Observable.fromEvent(document, 'click');
var powersOfTwo = clicks
 .map(e => 1)
 .expand(x => {
    console.log(`输入:${x}`);
    return Rx.Observable.of(2 * x).delay(1000)})
 .take(5);
powersOfTwo.subscribe(x => console.log(`输出:${x}`));

示例监听页面的点击事件，每次点击发送源值1，并对源值执行expand，expand的映射函数也是将源值乘以2，并延时1000毫秒。

点击一次，输出结果是和上面的例子一样。连续点击两次，输出结果：

输出:1
输入:1
输出:1
输入:1
输出:2
输入:2
输出:2
输入:2
输出:4
输入:4

由于例子里做了延时，多次点击的时间不同，输出的结果也是不同的。

递归调用的终止

默认情况下，expand递归调用时不限次数的，那如何终止呢。可以有以下一些可以终止递归调用：

1、使用take()操作符

2、映射函数project，返回一个empty Observable

var source = new Rx.Observable.of(3);

source.expand(function(x) {
  console.log('count: ' + x);
  x--;

return (x >= 0) ? Rx.Observable.just(x) : Rx.Observable.empty()
})
.subscribe(
    function (x) {
        console.log('Next: ' , x);
    },
    function (err) {
        console.log('Error: ' + err);   
    },
    function () {
        console.log('Completed');   
    });