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.

Answer 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.

Answer 2

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));