Is it possible to write an rxjs operator that controls subscription to it's source?

Question

Let's say I have an observable source that has the following properties:

It makes a network request the first time it's subscribed to
It's idempotent, so it will always emit the same value after it's first subscribed to
It's low priority, so we don6t6 want to be too eager in subscribing to it

What would be ideal is if there were some sort of operator delaySubscriptionUntil that would delay subscription until some other observable s emits a value.

So for example:

const s = new Subject<void>();
const l = source
  .pipe(
    delaySubscriptionUntil(s));

l.subscribe(console.log);
// The above won't print anything until this line executes
s.next();

I looked through the documentation to see if there's an existing operator like this, but haven't found one.

Adrian Brand · Answer 1 · 2023-02-06T21:10:54.830

1

You just put the subject first and switchMap

const l = s.pipe(
  switchMap(() => source)
);

Once the subject emits then the source will be subscribed to.

Any thing that is after wont work as it relies on the previous observable emitting a value. You can have a filter in the chain that stops the previous observable's emission being emitted but there is nothing you can pass back up the chain to control outer subscriptions.

You could use a takeWhile

let allow = false;

const l = source.pipe(
  takeWhile(allow)
);

but here the subscription to source is active, it is emitting values, they are just stopped from being passed through.

So you could make a similar operator that keeps an internal flag and is flipped by a subject but source is still going to be emitting, you are just filtering values. You could buffer up the values if you don't want to lose them.

edited Feb 06 '23 at 21:10

answered Feb 06 '23 at 20:50

Adrian Brand

20,384
4
39
60

The `takeWhile` solution wouldn't work because the idea would be to prevent the source observable from doing the side effects it has to do to emit any values. – Finley Baker Feb 06 '23 at 21:30
Which is why I have put it in there and explained how anything in the pipe chain relies on the source emitting a value before it is invoked. You will need to have the subject earlier in the chain not later. – Adrian Brand Feb 06 '23 at 22:08

score 0 · Answer 2 · answered Feb 15 '23 at 17:55

You could use share() which will share the result of anything that happened before it until you call sub.next() with a new url then the request will happen again.

const sub = new BehaviorSubject<string>('http://example.com/api');
const result$ = sub.pipe(
  exhaustMap(url => this.http.get(url)),
  share()
)

// Each one of these subscriptions will share the result.
// The http request will be called only once
// until you send a new url to the BehaviorSubject.
result$.subscribe(val => console.log(val));
result$.subscribe(val => console.log(val));
result$.subscribe(val => console.log(val));
result$.subscribe(val => console.log(val));

Is it possible to write an rxjs operator that controls subscription to it's source?

2 Answers2