RXJS система кеширования методом "next"

Question

Я могу построить систему кэширования с помощью BehaviourSubject:

class MessagesStore {
     constructor() {
         timer(0, 60 * 1000).pipe(
             switchMap(_ => this.messagesService.getMessages())
         ).subscribe(ms => this.subject.next(ms))
     }

     get messages$() {
         return this.subject.asObservable()
     }
}

Но для этого решения сообщения будут обновляться, когда какой-либо компонент не подписывается на сообщения $.Так что я могу построить кеширование без темы:

get messages$() {
    return timer(0, 60 * 1000).pipe(switchMap(_ => 
        this.messagesService.getMessages()), shareReplay(1))
}

readMessages() {
    this.subject.value.forEach(ms => ms.read = true)
    this.subject.next(subject.value)
}

Но теперь я не могу использовать метод "next"

Возможно ли иметь тему, но с автоматическим отписавшимся источником или чем-то подобным?

РЕДАКТИРОВАТЬ: Я придумал:

    let customUpdate: Subject<number> = new Subject();
    let lastNumber = 0;
    const fapy = interval(1000).pipe(
      tap(data => console.log('updater', data)), 
      merge(customUpdate),
      publishReplay(1),
      refCount(),
      tap(data => lastNumber = data)
    )
    const sub1 = fapy.subscribe(data => console.log('receirver1', data))
    const sub2 = fapy.subscribe(data => console.log('receirver2', data))
    let sub3

    setTimeout(_ => customUpdate.next(lastNumber + 1), 3000)
    setTimeout(_ => sub1.unsubscribe(), 3000)
    setTimeout(_ => sub2.unsubscribe(), 5000)
    setTimeout(_ => {sub3 = fapy.subscribe(data => console.log('receirver3', data))}, 7000)
    setTimeout(_ => {sub3.unsubscribe()}, 10000)

Я чувствую себя хозяином rxjs: d

Answer 1

Вы можете использовать библиотеку lodash, чтобы сравнить ваше текущее значение и только что введенное значение и обновить значение только , если они различаются.

Примерно так:

timer(0, 60 * 1000).pipe(switchMap(_ => this.messagesService.getMessages())
    .subscribe(ms => this.cache(this.subject, ms))

get messages$() {
    return this.subject.asObservable()
}

public cache(subject: BehaviorSubject<any>, next: any): void {
        if (!_.isEqual(subject.getValue(), next)) {
            subject.next(next);
        }
    }