响应式编程实战—— RxJS 改变事件流与合并事件流
今天我们来优化一下之前的程序。在 scan 中我们以匿名函数的形式对一个对象的属性了进行了加 1 操作，我们可以把这个匿名函数变成具名函数，这样做更加灵活，复用性也更佳，对吗？因此程序变成了这样：
const addOne = (acc) => ({count: acc.count + 1})
startBtnClick$
.pipe(
switchMapTo(intervalCanBeStopped$),
startWith({count:0}),
scan(addOne),
)
如果我们现在想在不改变从程序结构的情况下，点击开始按钮后计时器从 0 开始计数该怎么做？我们先来测试下：
const reset = (acc) => ({count: 0})
startBtnClick$
.pipe(
switchMapTo(intervalCanBeStopped$),
startWith({count:0}),
scan(reset),
)
当我们点击开始按钮时，会发现程序一直输出 0，说明重置生效了。到这里，我们总结出传递给 scan 不同的行为会有不同的结果。然而有什么办法可以不用我们手动拷贝粘贴，而是通过某个操作符来完成呢？
mapTo：这个操作符接收一个参数，将原事件流中的事件替换成这个参数。
const addOne = (acc) => ({count: acc.count + 1})
startBtnClick$
.pipe(
switchMapTo(intervalCanBeStopped$),
mapTo(addOne),
startWith({count:0}),
scan(/*todo*/),
)
通过 mapTo 操作符，我们把时间间隔事件流中的事件都变成了 addOne 函数，也就是说传递给 scan 的是一个函数，scan 中的累积函数该怎样写呢？
const addOne = (acc) => ({count: acc.count + 1})
startBtnClick$
.pipe(
switchMapTo(intervalCanBeStopped$),
mapTo(addOne),
startWith({count:0}),
scan((acc, curr) => curr(acc)),
)
我们解释一下。首先到达 scan 操作符的事件为 startWith 中的参数，也就是 {count: 0}，也就是说，累积函数第一次运行的返回值为 {count: 0}，这个返回值将作为下一次运行的 acc 参数。搞清楚这一点，我们再来看第二个到达 scan 操作符的事件是什么，很明显是 addOne 函数，累积函数中的 curr 参数将被赋值为这个函数。现在累积函数的参数都已经确定了，返回值该怎么写呢？很明显，把 acc 作为参数传递给 curr 函数，计算出返回值，也就是再下一次的 acc 的值，再下一次到来的还是 addOne 函数，如果不停止，将一直执行上面的操作，也就是加 1操作。这个 scan 做的事情像不像 redux 做的事情？
接下来，该把重置按钮加上了。Rx 编程模型中最有趣的事情来了，搭积木。我们该如何把 resetBtnClick$，也就是重置事件流和原来的事件流拼在一起。
我们知道加 1 操作是由时间间隔流变换而来的，重置按钮做的是清零操作，也就是说，重置事件流至少要放在 mapTo 做转换之前（先不考虑转换操作），然后 mapTo 根据到达的事件做判断，是加 1 还是重置，对么？那说明，重置流应该和时间间隔流应该属于同一个事件流。merge 操作符恰恰就是干这个的。
merge：通过查看官方文档，我们会发现 merge 操作符有多个重载实现。我们用到的是最基本的传给它多个事件流参数。
startBtnClick$
.pipe(
switchMapTo(merge(intervalCanBeStopped$, resetBtnClick$)),
mapTo(addOne),
startWith({ count: 0 }),
scan((acc, current) => current(acc))
)
实际运行效果肯定是不对的，因为根本就没有重置操作。点击重置按钮进行只是加 1 操作而已。事件流合并到了一起，如何区分事件呢？很简单：
startBtnClick$
.pipe(
switchMapTo(
merge(
intervalCanBeStopped$,
resetBtnClick$
)
),
map(v => {
if (typeof v === 'number') {
return addOne
} else {
return reset
}
}),
startWith({ count: 0 }),
scan((acc, current) => current(acc))
)
这样做确实可以实现我们需要的效果，但，我们仔细想一想，事件流中的事件对我们的作用只是用来区分行为，那么我们是不是可以在原始流就把各自的事件转换为各自的行为呢？当然可以，我觉得这才是 Rx 编程模型想让我们做的。
startBtnClick$
.pipe(
switchMapTo(
merge(
intervalCanBeStopped$.pipe(mapTo(addOne)),
resetBtnClick$.pipe(mapTo(reset))
)
),
startWith({ count: 0 }),
scan((acc, current) => current(acc))
)
在合并两个事件流之前，分别把两个事件流中的事件转换为了各自代表的行为，再合并为一个我们可以称之为行为事件流的东西。完整程序代码如下：
import React, { useRef, useEffect } from "react";
import { fromEvent, interval, merge } from "rxjs";
import { takeUntil, switchMapTo, scan, startWith, mapTo } from "rxjs/operators";
export default function App() {
const pauseBtnRef = useRef(null);
const startBtnRef = useRef(null);
const resetBtnRef = useRef(null);
const addOne = acc => ({ count: acc.count + 1 });
const reset = acc => ({ count: 0 });
useEffect(() => {
const pauseBtnClick$ = fromEvent(pauseBtnRef.current, "click");
const startBtnClick$ = fromEvent(startBtnRef.current, "click");
const resetBtnClick$ = fromEvent(resetBtnRef.current, "click");
const perSecond$ = interval(1000);
const intervalCanBeStopped$ = perSecond$.pipe(takeUntil(pauseBtnClick$));
const addOneOrReset$ = merge(
intervalCanBeStopped$.pipe(mapTo(addOne)),
resetBtnClick$.pipe(mapTo(reset))
)
const subscription = startBtnClick$
.pipe(
switchMapTo(
addOneOrReset$
),
startWith({ count: 0 }),
scan((acc, current) => current(acc))
)
.subscribe(v => console.log(v));
return () => {
subscription.unsubscribe();
};
});
return (
<div className="App">
<button ref={startBtnRef}>开始按钮</button>
<button ref={pauseBtnRef}>暂停按钮</button>
<button ref={resetBtnRef}>重置按钮</button>
</div>
);
}
有任何问题，请添加微信公众号“读一读我”。
