在mergeMap中取消重复订阅答案

【问题标题】：Cancel repeated subscription in mergeMap在mergeMap中取消重复订阅
【发布时间】：2021-04-03 09:52:12
【问题描述】：

如何结合distinct、switchMap和mergeMap操作符，使得当source发出重复值时（通过distinct.keySelector检测），取消之前的订阅（如switchMap中），但如果没有重复值则跟随mergeMap 的行为？

例子：

source = from(1, 2, 1, 2, 3) // 'abcde'
result = source.pipe(delay(), combination() // '--cde'

我目前正在做类似的事情：

const activeSubscriptions = new Map();
source$.pipe(
  mergeMap((value) => {
    const pendingSubscription = activeSubscriptions.get(value);
    if (pendingSubscription) {
      pendingSubscription.unsubscribe();
      activeSubscriptions.delete(value);
    }
    const request$ = new Subject();
    const subscription = this.service.get(value).subscribe({
      complete: () => request$.complete(),
      error: (err) => request$.error(err),
      next: (value) => request$.next(value),
    });
    activeSubscriptions.set(value, subscription);
    return request$;
  })
);

但正在寻找更好的方法来做到这一点。

提前谢谢你

【问题讨论】：

标签： rxjs rxjs5 rxjs-observables rxjs-pipeable-operators

【解决方案1】：

我认为您可以为此使用windowToggle 运算符：

src$ = src$.pipe(shareReplay(1));

src$.pipe(
  ignoreElements(),
  windowToggle(src$.pipe(observeOn(asyncScheduler)), openValue => src$.pipe(skip(1), filter(v => v === openValue))),
  mergeMap(
    window => window.pipe(
      startWith(null),
      withLatestFrom(src$.pipe(take(1))),
      map(([, windowVal]) => windowVal),
    )
  ),
)

observeOn(asyncScheduler) 的替代品也可以是delay(0)，重要的是要确保src$ 的订阅者收到值的顺序正确。在这种情况下，我们要确保当src$ 发出时，首先进行清理，这就是我们使用src$.pipe(observeOn(asyncScheduler)) 的原因。

ignoreElements() 被使用是因为每个窗口只与一个值配对，即创建窗口的那个值。传递给windowToggle 的第一个参数将描述可以创建windows 的可观察对象。所以，我们只需要这些，因为我们能够在

的帮助下获得最后一个值

window => window.pipe(
  startWith(null),
  withLatestFrom(src$.pipe(take(1))),
  map(([, windowVal]) => windowVal),
)

顺便说一句，window is nothing but a Subject。

最后，如果您想在window 的管道中执行异步操作，您必须确保在窗口完成（关闭）时取消订阅所有内容。为此，您可以尝试以下操作：

window => window.pipe(
  startWith(null),
  withLatestFrom(src$.pipe(take(1))),
  map(([, windowVal]) => windowVal),
  switchMap(val => /* some async action which uses `val` */),
  takeUntil(window.pipe(isEmpty()))
)

其中isEmpty 将在源（在本例中为window）完成时发出true 或false。 false 表示源在发出complete 通知之前已经发出了至少一个值，否则true。在这种情况下，我想说它是true 还是false 无关紧要，因为window 本身不会发出任何值（因为我们使用了ignoreElements，它会忽略除error 和@ 之外的所有内容987654348@ 通知）。

【讨论】：