switchMap与mergeMap结合

Question

我有一个可以观察到的位置，每个新值都应该导致一个HTTP请求。在客户端，我只关心最新的响应值；但是，为了监视/等目的，我希望完成每个请求。

我现在的情况是：

function simulate(x) {
  // Simulate an HTTP request.
  return of(x).pipe(delay(6));
}

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

当我将switchMap用于someMapFunc时，我会得到一组正确的响应(只有最新的)。但是，如果请求花费的时间太长，它将被取消。

当我使用mergeMap时，我得到了正确的请求集(每个请求都完成了)，但是我得到了错误的响应集(每个响应)。

​

​

有一种方法可以用mergeMap switchMap**?**的响应来获取的请求--我知道我可以将它写成一个自定义操作符，但是我想知道是否可以用现有的/标准的rxjs操作符来构建它。总结一下我的想法：

• 是switchMap的一个版本，它在切换时不取消订阅；
• 是mergeMap的一个版本，它只从最新的可观察到的内部发出值。

编辑：基于已接受的答案，我能够得到以下内容，这是可行的：

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

Answer 1

我不能百分之百确定这是不是你想要的，我也没有完全测试过，但是我创建了一个自定义操作符，它可以做一些与你所追求的接近的事情。也许你可以再多修一点。

这是一个过滤掉“旧”值的mergeMap。旧值是在较新的源开始排放后产生的源的排放。

function orderedMergeMap<T, R>(project: (v:T) => Observable<R>): OperatorFunction<T, R> {
  return s => defer(() => {
    let recent = 0;
    return s.pipe(
      map((v, i) => ({order: i, payload: v})),
      mergeMap(({order, payload}) => project(payload).pipe(
        map(v => ({order, payload: v}))
      )),
      tap(({order}) => {
        if(order > recent) recent = order;
      }),
      filter(({order}) => order < recent),
      map(({payload}) => payload)
    );
  });
}

该版本的OP确定如下：

function orderedMergeMap<T, R>(project: (v:T) => Observable<R>): OperatorFunction<T, R> {
  return s => defer(() => { 
    let recent = 0; 
    return s.pipe( 
      mergeMap((data, idx) => { 
        recent = idx; 
        return project(data).pipe(
          filter(() => idx === recent)
        ); 
      }) 
    ); 
  }); 
}

Answer 2

我相信你需要concatMap()和last()的结合。

concatMap不订阅下一个可观察到的，直到上一个完成。使用它，您将确保请求执行的顺序。从描述中可以看出，与switchMap不同，它不会取消以前的订阅并让它们完成订阅。

last在完成时发出从源发出的最后一个值。使用它，您将确保只有一个(最后)结果将传递给结果。

您的代码将如下所示：

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