javascript - 在 RxJS 6 中重置 ReplaySubject

Question

我有一个可过滤的“事件日志”，目前使用 ReplaySubject 实现(因为一些组件使用它并且它们可能在不同时间订阅)。

当用户更改过滤器设置时，会发出新请求，但结果会附加到 ReplaySubject 而不是替换它。

我想知道是否有办法更新 ReplaySubject 以仅使用类似 switchMap 的方式发送新项目？

否则，我可能需要使用返回所有事件条目数组的 BehaviorSubject 或重新创建 ReplaySubject 并通知用户(可能通过使用另一个可观察对象)取消订阅并重新订阅新的可观察对象。

Answer 1

如果您希望能够在订阅者明确取消订阅和重新订阅的情况下重置主题，您可以这样做:

import { Observable, Subject } from "rxjs";
import { startWith, switchMap } from "rxjs/operators";

function resettable<T>(factory: () => Subject<T>): {
  observable: Observable<T>,
  reset(): void,
  subject: Subject<T>
} {
  const resetter = new Subject<any>();
  const source = new Subject<T>();
  let destination = factory();
  let subscription = source.subscribe(destination);
  return {
    observable: resetter.asObservable().pipe(
      startWith(null),
      switchMap(() => destination)
    ),
    reset: () => {
      subscription.unsubscribe();
      destination = factory();
      subscription = source.subscribe(destination);
      resetter.next();
    },
    subject: source
  };
}

resettable 将返回一个包含以下内容的对象:

• 一个observable，可重置主题的订阅者应该订阅它；
• 一个 subject，您将对其调用 next、error 或 complete；和
• 一个 reset 函数，它将重置(内部)主题。

你会像这样使用它:

import { ReplaySubject } from "rxjs";
const { observable, reset, subject } = resettable(() => new ReplaySubject(3));
observable.subscribe(value => console.log(`a${value}`)); // a1, a2, a3, a4, a5, a6
subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);
observable.subscribe(value => console.log(`b${value}`)); // b2, b3, b4, b5, b6
reset();
observable.subscribe(value => console.log(`c${value}`)); // c5, c6
subject.next(5);
subject.next(6);