RxJs - Calculate & emit values only if there is a subscriber

Question

I want to create an observable that emits file additions/removal (via chokidar). I am able to do this by something like this:

Rx.Observable.create((subscriber) => {
  this.watcher = chokidar.watch(
     this.contentPath
  );
  this.watcher.on('addDir', () => { subscriber.next(); });
  this.watcher.on('unlinkDir', () => { subscriber.next(); });
});

What I want to do is, I want to stop for watching files, if there is no subscriber and start again when something subscribes to it. Something like this, but with RxJs:

class Notifier {
  constructor() {
    this.subscriberCount = 0;
  }

  subscribe(onNext, onError, complete) {
    this.subscriberCount++;
    if (this.subscriberCount === 1) {
      this.startInternalWatcher();
    }
    return () => {
      this.subscriberCount--;
      if (this.subscriberCount === 0) {
        this.stopInternalWatcher();
      }
    }
  }
}

// files are not watched
const n = new Notifier();

const s1 = n.subscribe(() => {}) // files are being wacthed
const s2 = n.subscribe(() => {}) // files are being wacthed
s1() // unsubscribed from 1, files are still watched.
s2() // unsubscribed from 2, files are not watched because no one is interested in.

I am new to RxJs so I might be missing some obvious solution. Is this possible?

Answer 1

You're on the right track. First up, if you return a function from the creator it will be called when the subscription is cancelled, so you can use that to destroy the watcher.

That should solve most of your problem, but if you want to ensure there's a maximum of one "watcher" at one time you can tack on refCount:

return Rx.Observable.create((subscriber) => {
  this.watcher = chokidar.watch(
     this.contentPath
  );
  this.watcher.on('addDir', () => { subscriber.next(); });
  this.watcher.on('unlinkDir', () => { subscriber.next(); });

  return () => this.watcher.off('addDir unlinkDir');
})
.publish()
.refCount();