javascript - switchMap 由 RxJs 中的属性区分

Question

比方说，我有一个 Action 流。每个 Action 都分配了一些 id。像这样:

const actions$ = of({ id: 1 }, { id: 2 }, { id: 1 });

现在，对于每个操作，我想在 switchMap 中执行一些逻辑:

actions$.pipe(switchMap(a => /* some cancellable logic */)).subscribe(...);

问题是每个发出的 Action 都会取消之前的“一些可取消逻辑”。

是否可以根据操作 id 取消“一些可取消的逻辑”，最好是运算符(operator)？像这样的东西:

actions$.pipe(switchMapBy('id', a => /*some cancellable logic */)).subscribe(...)

本质上，switchMap 的当前行为:
1. actions$ 发出 id #1。 switchMap 订阅嵌套的 observable。
2. actions$ 发出 id #2。 switchMap 取消订阅之前嵌套的可观察对象。订阅新的。
3. actions$ 发出 id #1。 switchMap 再次取消订阅先前嵌套的可观察对象。订阅新的。

预期行为:
1. actions$ 发出 id #1。 switchMap 订阅嵌套的 observable。
2. actions$ 发出 id #2。 switchMap 再次订阅嵌套的可观察对象(这次使用 #2)。 区别在于，它不会取消#1 中的那个。
3. actions$ 发出 id #1。 switchMap 取消订阅 #1 的嵌套可观察对象。再次订阅 #1。

Answer 1

这似乎是 mergeMap 运算符的一个用例。 switchMap 的用例是只维护一个内部订阅并取消以前的订阅，这不是你想要的。您想要多个内部订阅，并且希望它们在同一 ID 的新值出现时取消，因此实现一些自定义逻辑来做到这一点。

类似的东西:

action$.pipe(
  mergeMap(val => {
    return (/* your transform logic here */)
              .pipe(takeUntil(action$.pipe(filter(a => a.id === val.id)))); // cancel it when the same id comes back through, put this operator at the correct point in the chain
  })
)

您可以通过编写自定义运算符将它变成可重用的东西:

import { OperatorFunction, Observable, from } from 'rxjs';
import { takeUntil, filter, mergeMap } from 'rxjs/operators';

export function switchMapBy<T, R>(
  key: keyof T,
  mapFn: (val: T) => Observable<R> | Promise<R>
): OperatorFunction<T, R> {
  return input$ => input$.pipe(
    mergeMap(val => 
      from(mapFn(val)).pipe(
        takeUntil(input$.pipe(filter(i => i[key] === val[key])))
      )
    )
  );
}

并像这样使用它:

action$.pipe(
  switchMapBy('id', (val) => /* your transform logic here */)
);

这是它的 Blitz :https://stackblitz.com/edit/rxjs-x1g4vc?file=index.ts