rxjs 5 publishReplay refCount

publishReplay(x).refCount() в сочетании выполняет следующие действия:

• Он создает ReplaySubject, который воспроизводит до x выбросов. Если x не определен, он воспроизводит весь поток.
• Это делает эту ReplaySubject многоадресную передачу совместимой с помощью оператора refCount (). Это приводит к тому, что одновременные подписки получают одинаковые выбросы.

В вашем примере есть несколько проблем, которые затрудняют совместную работу всего этого. См. Следующий измененный фрагмент:

var state = 5
var realSource = Rx.Observable.create(observer =>  {
  console.log("creating expensive HTTP-based emission"); 
  observer.next(state++);
//  observer.complete();
  
  return () => {
    console.log('unsubscribing from source')
  }
});


var source = Rx.Observable.of('')
  .do(() => console.log('stream subscribed'))
  .ignoreElements()
  .concat(realSource)
.do(null, null, () => console.log('stream completed'))
.publishReplay()
.refCount()
;
    
subscription1 = source.subscribe({next: (v) => console.log('observerA: ' + v)});
subscription1.unsubscribe();
 
subscription2 = source.subscribe(v => console.log('observerB: ' + v));
subscription2.unsubscribe();
    
subscription3 = source.subscribe(v => console.log('observerC: ' + v));
subscription3.unsubscribe();
    
subscription4 = source.subscribe(v => console.log('observerD: ' + v));
 

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.1.0/Rx.js"></script>

При запуске этого фрагмента мы ясно видим, что он не генерирует повторяющиеся значения для Observer D, а фактически создает новые выбросы для каждой подписки. Почему?

Подписка на каждую подписку отменяется до того, как состоится следующая подписка. Это фактически заставляет refCount уменьшаться до нуля, многоадресная рассылка не выполняется.

Проблема заключается в том, что поток realSource не завершается. Поскольку мы не осуществляем многоадресную рассылку, следующий подписчик получает новый экземпляр realSource через ReplaySubject, а новые выбросы добавляются к предыдущим уже отправленным сообщениям.

Таким образом, чтобы избавить ваш поток от многократного вызова дорогостоящего HTTP-запроса, вам нужно завершить поток, чтобы publishReplay знал, что ему не нужно повторно подписываться.

Обычно: refCount означает, что поток является горячим / общим, пока есть хотя бы 1 подписчик, однако он сбрасывается / холодный, когда подписчиков нет.

Это означает, что если вы хотите быть абсолютно уверены в том, что ничего не выполняется более одного раза, вы не должны использовать refCount(), а просто connect поток, чтобы активировать его.

В качестве дополнительного примечания: если вы добавите observer.complete() после observer.next(5);, вы также получите ожидаемый результат.

Примечание: Вам действительно нужно создать здесь свой собственный Obervable? В 95% случаев существующих операторов достаточно для данного варианта использования.

Это происходит потому, что вы используете publishReplay(). Он внутренне создает экземпляр ReplaySubject, в котором хранятся все проходящие значения.

Поскольку вы используете Observable.create, где вы испускаете одно значение, то каждый раз, когда вы вызываете source.subscribe(...), вы добавляете одно значение в буфер в ReplaySubject.

Вы не получаете call в начале каждой строки, потому что это ReplaySubject, который сначала излучает свой буфер, когда вы подписываетесь, а затем он подписывается на свой источник:

Подробнее о реализации см .:

• https://github.com/ReactiveX/rxjs/blob/master/src/operator/multicast.ts#L63

• https://github.com/ReactiveX/rxjs/blob/master/src/ReplaySubject.ts#L54

То же самое и при использовании publishReplay(1). Сначала он излучает буферизованный элемент из ReplaySubject, а затем еще один элемент из observer.next(5);