基于RxJs属性的switchMap

Question

比如说，我有一连串的行动。每个操作都分配了一些id。如下所示：

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

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

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

问题是，每个发出的操作都取消了以前的“一些可取消的逻辑”。

是否可以根据操作id取消“一些可取消的逻辑”，最好是一个运算符？类似于：

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

本质上，当前行为与 switchMap

1. 操作$ emits #1. switchMap订阅嵌套可观测的。
2. 操作$发出id #2. switchMap取消了以前嵌套可观察到的订阅。订阅新的。
3. 操作$发出id #1. switchMap再次取消以前嵌套的可观察到的订阅。订阅新的。

预期行为

1. 操作$ emits #1. switchMap订阅嵌套可观测的。
2. 操作$ emits id #2. switchMap再次订阅嵌套可观测(这次是#2)。和这里的不同之处在于，它不会取消#1中的一个。
3. 操作$ emits #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 */)
);

这是一次闪电战：https://stackblitz.com/edit/rxjs-x1g4vc?file=index.ts

Answer 2

使用filter操作在switchMap之前排除取消的ids，如下所示

of({ id: 1 }, { id: 2 }, { id: 1 }).pipe(
   filter(id => ![1,2].includes(id)), // to exclude ids (1,2)
   switchMap(id => /*some cancellable logic */ )
).subscribe(...)