RxJS: Welcome on Board Upgraded ShareReplay Operator

👀 Background

Are you well-versed with RxJS? That’s what I thought until I’d come across the opened issue on Github about the RxJS shareReplay functionality last February. It described by Ben Lash, however, from the very beginning, it was unclear how precisely it’s supposed to operate in different scenarios.

👂 The shareReplay operator behavior

Last year it has been discussed a lot on careful usage of the shareReplay operator. What could be the reason? After all, most of us “know” how and in what case to use it, right?

Just to remind you (or in case you hear about it the very first time) let’s consider the following code:

const source$ = interval(1000).pipe(
  mapTo('I\'m from the shareReplay subscription'),
  tap(console.log),
  shareReplay(1)
)
const internalSubscription = source$.pipe(
  mapTo('I\'m from the source$ subscription')
).subscribe(console.log);
setTimeout(() => {
  internalSubscription.unsubscribe();
}, 2000);

I bet that your first expectations from the code above are the same as were mine. As we know the shareReplay operator underway uses the ReplaySubject and the refCount variable that counts internal subscriptions. And when the refCount drops to zero due to unsubscription - it should end the subscription and not recycle the ReplaySubject, right?

I mean that as soon as we unsubscribe from the source the interval observable will stop emitting new values... Unfortunately, my expectations got shattered, because the shareReplay does never end the subscription under the hood and the interval observable keeps emitting values after we stopped listening.

You will see the next output:

// 1st emit
I’m from the shareReplay subscription
I’m from the source$ subscription
// 2nd emit
I’m from the shareReplay subscription
I’m from the source$ subscription
// 3rd emit
I’m from the shareReplay subscription
// 4th emit
I’m from the shareReplay subscription
...

Are you surprised? Yeah, it’s totally unexpected behavior that within the specific scenario, such as endless streaming data, might lead to tremendous memory leaks or even some bugs in your app. What will be the rescue?🤔

💪 Merged PR and upgraded functionality

After six months of heated discussions inside the community it has been decided to open a pull request that suggests to add the config parameter (an object) to the shareReplay operator. Here it is:

export interface ShareReplayConfig {
  bufferSize?: number;
  windowTime?: number;
  refCount: boolean;
  scheduler?: SchedulerLike;
}

More specifically, the option refCount can be used to control the behavior. In case you set the following configuration { refCount: true }, then the subscription to the source$ would be a reference-counted and obviously when you unsubscribe from the source$, the refCount will go from 1 to 0. By default, it's not and you’d get the same functionality as before. Check out the code example below:

const source$ = interval(1000).pipe(
  mapTo('I\'m from the shareReplay subscription'),
  tap(console.log),
  shareReplay({bufferSize: 1, refCount: true})
)
const internalSubscription = source$.pipe(
  mapTo('I\'m from the source$ subscription')
).subscribe(console.log);
setTimeout(() => {
  internalSubscription.unsubscribe();
}, 2000);
// When executed, we see in the console:
// 1st emit
I’m from the shareReplay subscription
I’m from the source$ subscription
// 2nd emit
I’m from the shareReplay subscription
I’m from the source$ subscription

Finally, the PR was merged and deployed within the latest version of RxJS 6.4.0 on January 30th, 2019.

👌 Conclusion

All in all, in case you hear about the shareReplay issue for the first time and you have probably used it for endless streaming data, you may have a bad feeling now because your apps might be in trouble. Especially if you don’t have an option to update your current RxJS version to the latest one and pass the new config to the shareReplay.

At least, as long as you don’t have such a possibility and you want your apps to work stably, you should choose the safer option - use publishReplay and refCount to avoid possible memory leaks. Otherwise, update your RxJS, implement the new config where it's necessary and finally you’d get a chance to sleep well!

Liked that? We’ve done our best! Go to our blog to find more useful articles.