Сброс ReplaySubject в RxJS 6

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, на которые должны подписаться подписчики переустанавливаемого субъекта;
• a subject, на который вы бы позвонили next, error или complete; и
• a 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);