在 rxjs 中，如何从由 takeUntil 结束的 switchMap 中获取最后一个值？答案

【问题标题】：In rxjs, how do I get the last value from a switchMap that is ended by takeUntil?在 rxjs 中，如何从由 takeUntil 结束的 switchMap 中获取最后一个值？
【发布时间】：2021-02-07 18:47:49
【问题描述】：

这是一个典型的 rxjs 拖放示例：

const drag$ = mousedown$.pipe(
  switchMap(
    (start) => {
      return mousemove$.pipe(map(move => {
        move.preventDefault();
        return {
          left: move.clientX - start.offsetX,
          top: move.clientY - start.offsetY
        }
      }),
        takeUntil(mouseup$));
    }));

drag$.subscribe(pos => {
  box.style.top = `${pos.top}px`
  box.style.left = `${pos.left}px`
});

takeUntil 将取消订阅 mouseMove$ 的内部 observable。但是，如果我需要内部 observable 中的最后一个值以便在 mouseUp 上对其进行处理呢？我知道我可能可以为全局临时变量分配一个值来实现它，但我试图避免这种情况。

我正在尝试做的更复杂的故事是这样的：

我正在开发一个 3D 游戏，我想在其中实现抓斗和投掷。我已将其设置为类似于 rxjs 锥形拖放场景，在该场景中我订阅了控制器“挤压”事件，然后 switchMaping 到另一个可观察到的控制器移动，在该场景中我“拖动”并重新定位对象我的手在哪里，最后我takeUntil '挤压'被释放，放开物体。

但是我还需要实现一个“投掷”物体的部分，即我需要在释放挤压时手部运动的速度矢量向物体施加冲力，使其飞走当我放手的时候。可以使用scan 在switchMap 的pipe 内的每一帧上计算该向量。

问题是takeUntil 只是停止数据流，我无法访问scan 的累加器中的最后一个值。

【问题讨论】：

标签： rxjs

【解决方案1】：

您可以使用withLatestFrom() 运算符来获取另一个可观察对象的最新发射。结果是一个流值数组，后面是另一个 observable 的最后一个发射：

mouseup$.pipe(
  withLatestFrom(drag$)
).subscribe(
  ([mouseup, drag]) => console.log({drag, mouseup})
);

这是一个StackBlitz 示例。

【讨论】：

【解决方案2】：

这是一个非常有趣的问题，我猜你还需要 drag 事件发生时的值，以及在 mouseup 事件发生之前它最后发出的值。

我认为一种方法是：

const drag$ = mousedown$.pipe(
  switchMap(
    (start) => {
      mousemove$ = mousemove$.pipe(
        map(move => {
          move.preventDefault();
          return {
            left: move.clientX - start.offsetX,
            top: move.clientY - start.offsetY
          }
        }),
        takeUntil(mouseup$),
        share(),
      );

      return concat(
        // get the `mousemove` events as they take place
        mousemove$,

        mousemove$.pipe(
          reduce((acc, crt) => { /* ... */ }),
        ),
      )
}));

reduce 运算符的工作方式与scan 相同，只是前者将在源完成时发出累积值。

如果您想在两个 observable 中使用相同的 scan 逻辑，那么 switchMap 的内部 observable 可能如下所示：

mousemove$ = mousemove$.pipe(
  map(move => {
    move.preventDefault();
    return {
      left: move.clientX - start.offsetX,
      top: move.clientY - start.offsetY
    }
  }),
  scan((acc, crt) => { /* ... */ }),
  takeUntil(mouseup$),
  share(),
);

return concat(
  mousemove$,

  mousemove$.pipe(
    // it will emit the last emitted value when the source completes
    takeLast(1),
  ),
)

【讨论】：