Replay

Также существует оператор shareReplay, который отслеживает количество подписчиков и отключается от исходного Observable, когда это число падает до нуля. Оператор shareReplay принимает три необязательных параметра и возвращает обычный Observable:

bufferSize

максимальное количество элементов для буферизации и повторной отправки последующим подписчикам

window

возраст в миллисекундах, по истечении которого элементы в этом буфере могут быть удалены без отправки последующим подписчикам

scheduler

Scheduler, на котором будет работать этот оператор

Пример кода

var interval = Rx.Observable.interval(1000);

var source = interval
    .take(4)
    .doAction(function (x) {
        console.log('Side effect');
    });

var published = source
    .shareReplay(3);

published.subscribe(createObserver('SourceA'));
published.subscribe(createObserver('SourceB'));

// Creating a third subscription after the previous two subscriptions have
// completed. Notice that no side effects result from this subscription,
// because the notifications are cached and replayed.
Rx.Observable
    .return(true)
    .delay(6000)
    .flatMap(published)
    .subscribe(createObserver('SourceC'));

function createObserver(tag) {
    return Rx.Observer.create(
        function (x) { console.log('Next: ' + tag + x); },
        function (err) { console.log('Error: ' + err); },
        function () { console.log('Completed'); });
}

Side effect
Next: SourceA0
Next: SourceB0
Side effect
Next: SourceA1
Next: SourceB1
Side effect
Next: SourceA2
Next: SourceB2
Side effect
Next: SourceA3
Next: SourceB3
Completed
Completed
Next: SourceC1
Next: SourceC2
Next: SourceC3
Completed

Операторы replay и shareReplay находятся в следующих дистрибутивах:

• rx.all.js
• rx.all.compat.js
• rx.binding.js (требует rx.js или rx.compat.js)
• rx.lite.js
• rx.lite.compat.js