CODEWITHSUNDEEP
rxjsobservablereactive-programming
maxime1992
maxime1992

Reputation: 23813

Rxjs refCount callback to cleanup once every subscribers have unsubscribed?

I've got an observable which is not long lived (http request).

I'm using publishReplay(1) and refCount() so that when there an attempt to access it at the same time, it'll return the same value without triggering the http call again.

But if all the subscriptions are unsubscribed, I need to make some cleanup.

I can't use finalize because:

So basically what I'd like would be to pass a callback to refCount and call that callback when the number of subscriptions reaches 0. But it doesn't work like that. Is there any way to be "warned" when all the subscribers have unsubscribed?

The simplest way I can think of right now would be to create a custom operator that'd pretty much extend refCount to add a callback.

Any better thoughts? I'm pretty sure that there's a better way of doing that.

Thanks!

Upvotes: 3

Views: 501

Answers (2)

Bielik
Bielik

Reputation: 992

I am not gonna lie, I was happy to find out I wasn't the only one looking for something like that. There is one another person.

I ended up doing something like that:

import { Observable } from 'rxjs';

export function tapTeardown(teardownLogic: () => void) {
  return <T>(source: Observable<T>): Observable<T> =>
    new Observable<T>((observer) => {
      const subscription = source.subscribe(observer);

      return () => {
        subscription.unsubscribe();
        teardownLogic();
      };
    });
}

And you use it like:

    const augmented = connection.pipe(
      tapTeardown(() => /* SOME TEARDOWN LOGIC */),
      shareReplay({ bufferSize: 1, refCount: true }),
    );

Upvotes: 1

maxime1992
maxime1992

Reputation: 23813

I've tried it and it seems to work correctly.

Here's how it's used:

import { of, timer } from 'rxjs'; 
import { map, publishReplay, take } from 'rxjs/operators';
import { refCountCb } from './refCountCb';

const source = timer(2000, 10000).pipe(
  map(x => `Hello ${x}!`),
  publishReplay(1),
  refCountCb(() => console.log('MAIN CLOSED'))
);

source.pipe(take(1)).subscribe(x => console.log(x));
source.pipe(take(1)).subscribe(x => console.log(x));

Output:

Hello 0!
Hello 0!
MAIN CLOSED

I've built the custom refCountCb operator based on the source of refCount. It's basically just adding a callback so I won't copy paste the whole code here but it's available on the stackblitz.

Full demo: https://stackblitz.com/edit/rxjs-h7dbfc?file=index.ts

If you have any other idea please share it, I'd be glad to discover different solutions!

Upvotes: 0

Related Questions