RxJS - How to share the output of an expensive observable but rerun that observable if its requested again after N seconds?

Question

Let's say we have this global const:

const isSignedIn = fromPromise(fetch('/api/is-signed-in'))
    .pipe(throttleTime(1000), shareReply(1));

After page load, several components will subscribe to this at the same time:

isSignedIn.subscribe(() => console.log('do 1st'));
isSignedIn.subscribe(() => console.log('do 2nd'));
isSignedIn.subscribe(() => console.log('do 3rd'));

The above will only call the API once, however i need it to call the API again (ie after 1 second) if another component subscribes to it.

isSignedIn.subscribe(() => console.log('button press'));

How do i that using RxJS?

Answer 1

I think this is what you want:

A pipeable operator (declare globally somewhere and import it)

 export const refreshAfter = (duration: number) => (source: Observable<any>) =>
                             source.pipe(
                                   repeatWhen(obs => obs.pipe(delay(duration))),
                                   publishReplay(1), 
                                   refCount());

Then use it like this:

data$ = fetch('/api/is-signed-in').pipe(refreshAfter(5000));   // refresh after 5000 ms

---

Note: You actually asked for this:

i need it to call the API again (ie after 1 second) if another component subscribes to it.

Not quite sure this is what you really meant. I think what you really meant was - you want the data to be refreshed for all components currently subscribed after an expiry time. Anyway my answer sends the new value to all listeners. If you really want what you originally said you'd need to add some kind of alternative repeat trigger.

But if this is for a global constant - the above is what I'm using for the same scenario.

Note: I haven't actually tested the handling of an error condition when the item is repested, but I think the error will propagate to all listeners.