RxJs

Question

我想创建一个发出文件 additions/removal 的可观察对象（通过 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

你走在正确的轨道上。首先，如果您 return 来自创建者 it will be called when the subscription is cancelled 的函数，那么您可以使用它来破坏观察者。

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

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

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

rxjs5