RxJS中如何结合mergeMap全请求完成与switchMap仅返回最新响应的特性

需求描述 #

我有一个Observable，每个新值触发一次HTTP请求：

• 客户端业务逻辑仅需要使用最新请求的响应结果
• 出于监控、埋点等需求，所有已经发起的请求必须完整执行完毕，不能被取消

当前实现与问题 #

现有代码如下：

function simulate(x) {
  // 模拟HTTP请求
  return of(x).pipe(delay(6));
}

source$.pipe(
  someMapFunc(x => simulate(x)),
);

目前尝试两种内置操作符都不能满足需求：

• 使用switchMap：仅返回最新请求的响应，符合业务侧数据要求，但新请求发起时会取消未完成的旧请求，不满足监控需求
• 使用mergeMap：所有请求都会完整执行，满足监控需求，但会返回所有请求的响应结果，不符合仅需最新数据的业务要求

对应弹珠逻辑示意图：

核心诉求 #

有没有办法仅通过RxJS标准操作符组合，同时实现两种操作符的优势？即：

• 切换新请求时不取消未完成的旧请求（mergeMap的特性）
• 最终仅发射最新请求的响应结果（switchMap的特性）

最终实现方案 #

通过内置操作符组合封装的自定义操作符实现如下：

function orderedMergeMap(project) {
  return (s) => defer(() => {
    let recent = 0;
    return s.pipe(
      mergeMap((data, idx) => {
        recent = idx;
        return project(data).pipe(filter(() => idx === recent));
      })
    );
  });
}

内容的提问来源于stack exchange，提问作者Steve D