RxJs - 仅在有订阅者时才计算和发出值

Question

我想创建一个发出文件添加/删除的可观察对象(通过 chokidar)。我可以通过这样的方式做到这一点:

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

我想做的是，我想停止观看文件，如果没有订阅者，并在有人订阅它时重新开始。像这样的东西，但是使用 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.

我是 RxJs 的新手，所以我可能会遗漏一些明显的解决方案。这可能吗？

Answer 1

您走在正确的轨道上。首先，如果您从创建者那里返回一个函数 it will be called when the subscription is cancelled ，所以你可以用它来摧毁观察者。

这应该可以解决你的大部分问题，但如果你想确保一次最多有一个“观察者”，你可以添加 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();