Как создать Observable, что срабатывает только когда она имеет абонентов и предоставляет последнее значение для новых абонентов сразу

Question

Я пытаюсь создать поток/наблюдаемым, что ...

1. только выводит события, когда он имеет подписчиков
2. Предоставляет новых подписчиков с последним значением.

Конкретный случай заключается в том, что мне нужно наблюдение, которое вызывает вызов API Async всякий раз, когда происходит конкретное событие, но только если у него есть подписчики. Я пытаюсь избежать ненужных вызовов API.

мне удалось создать поток, который срабатывает только когда он имеет подписчиков, как это ...

let dataStream = Rx.Observable 
    .interval(1000) // Fire an event every second 
    .singleInstance() // Only do something when we have subscribers 
    .startWith(null) // kick start as soon as something subscribes 
    .flatMapLatest(interval => SomeAPI.someDataGet()) // get data, returns a promise 

И это работает. Если I console.log(...) в методе SomeAPI.someDataGet, я вижу, что он стреляет, когда у потока есть подписчики. И моя реализация выглядит очень хорошо, потому что я делаю это, чтобы подписаться и отказаться от подписки, которая очень хорошо вписывается в методы жизненного цикла компонентов React.

let sub1; 
sub1 = dataStream.subscribe(x => console.log('sub1', x)); 
sub1.dispose(); 

Я также хочу, чтобы новые подписчики получили самую последнюю ценность в тот момент, когда они подписываются. Здесь я боюсь. Если я сделаю это ...

let sub1, sub2; 
sub1 = dataStream.subscribe(x => console.log('sub1', x)); 

setTimeout(() => { 
    sub2 = dataStream.subscribe(x => console.log('sub2', x)); 
}, 1500) 

... Я не вижу console.log для sub2 до следующего интервала.

Если мое понимание верное. Мне нужен Hot Observable. Так что я попытался создать поток, как это ...

let dataStream = Rx.Observable 
    .interval(1000) // Fire an event every second 
    .singleInstance() // Only do something when we have subscribers 
    .startWith(null) // kick start as soon as something subscribes 
    .flatMapLatest(interval => SomeAPI.someDataGet()) // get data 
    .publish() // Make this a hot observable; 

который, как я понимаю, должен сделать dataStreamhot observable.

Однако в моих тестах вторая подписка по-прежнему не принимает данные до следующего интервала. Кроме того, это приведет к необходимости подключения и отключения DataStream при подписке, что я бы хотел избежать, если это возможно.

Я совершенно новый для RxJS, и я не удивлюсь, если я неправильно понял, что здесь происходит.

Answer 1

Вместо .publish(), используйте .shareReplay(1).