CODEWITHSUNDEEP
rxjsrxjs5
Oleg Gello
Oleg Gello

Reputation: 293

rxjs 5 publishReplay refCount

I can't figure out how publishReplay().refCount() works.

For example (https://jsfiddle.net/7o3a45L1/):

var source = Rx.Observable.create(observer =>  {
  console.log("call"); 
  // expensive http request
  observer.next(5);
}).publishReplay().refCount();

subscription1 = source.subscribe({next: (v) => console.log('observerA: ' + v)});
subscription1.unsubscribe();
console.log(""); 

subscription2 = source.subscribe({next: (v) => console.log('observerB: ' + v)});
subscription2.unsubscribe();
console.log(""); 

subscription3 = source.subscribe({next: (v) => console.log('observerC: ' + v)});
subscription3.unsubscribe();
console.log(""); 

subscription4 = source.subscribe({next: (v) => console.log('observerD: ' + v)});
subscription4.unsubscribe();

gives the following result:

call observerA: 5

observerB: 5 call observerB: 5

observerC: 5 observerC: 5 call observerC: 5

observerD: 5 observerD: 5 observerD: 5 call observerD: 5

1) Why are observerB, C and D called multiple times?

2) Why "call" is printed on each line and not in the beginning of the line?

Also, if i call publishReplay(1).refCount(), it calls observerB, C and D 2 times each.

What i expect is that every new observer receives the value 5 exactly once and "call" is printed only once.

Upvotes: 23

Views: 24869

Answers (3)

Mark van Straten
Mark van Straten

Reputation: 9425

publishReplay(x).refCount() combined does the following:

  • It create a ReplaySubject which replay up to x emissions. If x is not defined then it replays the complete stream.
  • It makes this ReplaySubject multicast compatible using a refCount() operator. This results in concurrent subscriptions receiving the same emissions.

Your example contains a few issues clouding how it all works together. See the following revised snippet:

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>

When running this snippet we can see clearly that it is not emitting duplicate values for Observer D, it is in fact creating new emissions for every subscription. How come?

Every subscription is unsubscribed before the next subscription takes place. This effectively makes the refCount decrease back to zero, no multicasting is being done.

The issue resides in the fact that the realSource stream does not complete. Because we are not multicasting the next subscriber gets a fresh instance of realSource through the ReplaySubject and the new emissions are prepended with the previous already emitted emissions.

So to fix your stream from invoking the expensive HTTP request multiple times you have to complete the stream so the publishReplay knows it does not need to re-subscribe.

Upvotes: 30

Olaf Horstmann
Olaf Horstmann

Reputation: 16882

Generally: The refCount means, that the stream is hot/shared as long as there is at least 1 subscriber - however, it is being reset/cold when there are no subscribers.

This means if you want to be absolutely sure that nothing is executed more than once, you should not use refCount() but simply connect the stream to set it hot.

As an additional note: If you add an observer.complete() after the observer.next(5); you will also get the result you expected.

Sidenote: Do you really need to create your own custom Obervable here? In 95% of the cases the existing operators are sufficient for the given usecase.

Upvotes: 10

martin
martin

Reputation: 96891

This happens because you're using publishReplay(). It internally creates an instance of ReplaySubject that stores all values that go through.

Since you're using Observable.create where you emit a single value then every time you call source.subscribe(...) you append one value to the buffer in ReplaySubject.

You're not getting call printed at the beginning of each line because it's the ReplaySubject who emits its buffer first when you subscribe and then it subscribes itself to its source:

For implementation details see:

The same applies when using publishReplay(1). First it emits the buffered item from ReplaySubject and then yet another item from observer.next(5);

Upvotes: 5

Related Questions