响应式编程实战——RxJS 重构组合事件流

今天我们来看看如何改变事件发生的频率。让我们添加两个按钮，分别代表了二分之一秒产生一个事件的事件流和四分之一秒产生一个事件的事件流。根据昨天的文章，我们可以想到用 merge 操作符把它们和开始按钮合并起来。

const time$ = merge(startBtnClick$, halfBtnClick$, quarterBtnClick$);

const subscription = time$
.pipe(
  switchMapTo(addOneOrReset$),
  startWith({ count: 0 }),
  scan((acc, current) => current(acc))
)

很明显，这三个按钮实现的效果是一样的，还是每隔一秒产生一个事件。既然我们要修改事件产生的频率，那么我们就要知道原来这个事件流是怎么定义的：

const perSecond$ = interval(1000);

我们看到是由 interval 的参数来控制产生事件的频率，来修改一下代码：

const time$ = merge(
  startBtnClick$.pipe(mapTo(1000)),
  halfBtnClick$.pipe(mapTo(500)),
  quarterBtnClick$.pipe(mapTo(250))
);

这样就行了吗？开玩笑，当然不行了，这只是把三个按钮的 click 事件产生的值做了修改。我们需要做的是修改传给 interval 的参数。现在的代码是这样的：

const perSecond$ = interval(1000);
const intervalCanBeStopped$ = perSecond$.pipe(takeUntil(pauseBtnClick$));
const addOneOrReset$ = merge(
  intervalCanBeStopped$.pipe(mapTo(addOne)),
  resetBtnClick$.pipe(mapTo(reset))
);

const time$ = merge(
  startBtnClick$.pipe(mapTo(1000)),
  halfBtnClick$.pipe(mapTo(500)),
  quarterBtnClick$.pipe(mapTo(250))
);

const subscription = time$
.pipe(
  switchMapTo(addOneOrReset$),
  startWith({ count: 0 }),
  scan((acc, current) => current(acc))
)
.subscribe(v => setTxt(v.count));

首先，我们得知，三个按钮点击后的事件流转到了 switchMapTo 操作符；其次，我们知道 addOneOrReset$ 是由 interval(1000) 组合而来的。也就是说，我们需要把流转到 switchMapTo 的事件传递给 addOneOrReset$。这里要用到了 switchMapTo 的兄弟，switchMap 操作符。

switchMap：参数为函数，这个函数接收事件流中的事件作为参数，返回值为另一个事件流。

让我们一步一步来修改代码，首先把时间参数传递给 switchMap 的函数参数：

const subscription = time$
.pipe(
  switchMap((time) => addOneOrReset$),
  startWith({ count: 0 }),
  scan((acc, current) => current(acc))
)
.subscribe(v => setTxt(v.count));

其次，我们需要把 addOneOrReset$ 拆开找到 interval 操作符，把 time 传递给它：

const subscription = time$
      .pipe(
        switchMap(time =>
          merge(
            interval(time).pipe(
              takeUntil(pauseBtnClick$),
              mapTo(addOne)
            ),
            resetBtnClick$.pipe(mapTo(reset))
          )
        ),
        startWith({ count: 0 }),
        scan((acc, current) => current(acc))
      )
      .subscribe(v => setTxt(v.count));

让我们来梳理一下流程：

1. 开始按钮（1秒按钮），1/2秒按钮，1/4秒按钮的点击事件合并为一个事件流，事件流中的事件为三个值 1000，500，250（谁点击就产生对应的数字）。
2. 时间数字来到了 switchMap 操作符，并作为输入参数传递给了 switchMap 的函数参数。
3. switchMap 的函数参数返回一个新的事件流，也就是我们之前的 addOneOrReset$。我们得把这个“积木”拆开得到 interval，并把时间数字传递给它。

最后我们可以把原来的 addOneOrReset$ 事件流改造一下：

const addOneOrReset = (time = 1000) =>
      merge(
        interval(time).pipe(
          takeUntil(pauseBtnClick$),
          mapTo(addOne)
        ),
        resetBtnClick$.pipe(mapTo(reset))
      );

把它改造为一个可以设置时间参数并带有默认值的积木。下面是最终实现完整代码：

import React, { useRef, useEffect, useState } from "react";

import { fromEvent, interval, merge } from "rxjs";
import { takeUntil, switchMap, scan, startWith, mapTo } from "rxjs/operators";

export default function App() {
  const [txt, setTxt] = useState("");

  const pauseBtnRef = useRef(null);
  const startBtnRef = useRef(null);
  const resetBtnRef = useRef(null);
  const halfBtnRef = useRef(null);
  const quarterBtnRef = 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 halfBtnClick$ = fromEvent(halfBtnRef.current, "click");
    const quarterBtnClick$ = fromEvent(quarterBtnRef.current, "click");

    const addOneOrReset = (time = 1000) =>
      merge(
        interval(time).pipe(
          takeUntil(pauseBtnClick$),
          mapTo(addOne)
        ),
        resetBtnClick$.pipe(mapTo(reset))
      );
    const time$ = merge(
      startBtnClick$.pipe(mapTo(1000)),
      halfBtnClick$.pipe(mapTo(500)),
      quarterBtnClick$.pipe(mapTo(250))
    );

    const subscription = time$
      .pipe(
        switchMap(addOneOrReset),
        startWith({ count: 0 }),
        scan((acc, current) => current(acc))
      )
      .subscribe(v => setTxt(v.count));

    return () => {
      subscription.unsubscribe();
    };
  }, []);

  return (
    <div className="App">
      <div style={{ fontSize: "30px" }}>{txt}</div>
      <button ref={startBtnRef}>开始</button>
      <button ref={pauseBtnRef}>暂停</button>
      <button ref={resetBtnRef}>重置</button>
      <button ref={halfBtnRef}>1/2秒</button>
      <button ref={quarterBtnRef}>1/4秒</button>
    </div>
  );
}

好了，如有任何问题，请添加微信公众号“读一读我”。