switchMap 结合 mergeMap

switchMap combined with mergeMap

我有一个 Observable,其中每个新值都应引起 HTTP 请求。在客户端,我只关心最新的响应值;但是,我希望每个请求都能在 monitoring/etc 内完成。目的。

我目前拥有的是这样的:

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

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

当我对 someMapFunc 使用 switchMap 时,我得到了正确的回复集(只有最新的)。但是,如果请求花费的时间太长,它将被取消。

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

有没有办法得到 mergeMap 的请求和 switchMap 的响应? 我知道我可以把它写成自定义运算符,但我想知道我是否可以用 existing/standard rxjs 运算符构建它。总结一下我的想法:

编辑:根据接受的答案,我得到了以下有效的答案:

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

我认为您需要 concatMap()last() 的组合。

concatMap 在前一个完成之前不会订阅下一个可观察对象。使用它您将确保请求执行的顺序。从描述中可以看出,它不会取消以前的订阅并让它们完成,这与 switchMap.

不同

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

您的代码将如下所示:

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

我不是 100% 确定这是否是您所追求的,而且我还没有对此进行全面测试,但我创建了一个自定义运算符,它可能会做一些接近您所追求的事情。也许你可以再修改一下。

这是一个过滤掉“旧”值的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)
        ); 
      }) 
    ); 
  }); 
}