Share operator that doesn't unsubscribe

Question

I need to lazy load some infinite streams because they are expensive to start. And I also don't ever want to stop them once they are started for the same reason.

I'm thinking it would be neat if there was a share operator that didn't unsubscribe from the underlying stream ever once it is subscribed for the first time, even when all downstream subscribers unsubscribe.

Right now I'm doing it with a publish and a connect on two different lines, which works alright but just seems clunky and not very rxjs like:

public data$(): Observable<any> {
    if (this.lazyData$) {
        return this.lazyData$;
    }

    this.lazyData$ = this.someDataProvider.data$()
    .publishReplay(1);

    this.lazyData$.connect();

    return this.lazyData$;
}

Also I want it to replay the last message to new subscribers as you see :)

The current implementation of shareReplay effects the behaviour you are after. It unsubscribes only when/if the source completes. See github.com/ReactiveX/rxjs/pull/2910 — cartant
Oh there we go, that worked thanks :) And thanks for the link too. I had to update to rxjs 5.5.5, but otherwise it just worked. You should put that as an answer so I can mark it as the accepted answer. — Sammi

cartant cartant · Accepted Answer · 2017-12-13T19:18:10

The shareReplay operator was added in RxJS version 5.4.0. And, in version 5.5.0 a bug was fixed so that it maintains its history when its subscriber count drops to zero.

With the fix, shareReplay will effect the behaviour you are looking for, as it will now unsubscribe from the source only when the source completes or errors. When the number of subscribers to the shared observable drops to zero, the shared observable will remain subscribed to the source.

The behaviour of shareReplay has changed several times and a summary of the changes - and the reasons for them - can be found in this blog post.

Share operator that doesn't unsubscribe

1 Answers