发布 observable 使其*热*

Question

我正在尝试创建一个不需要订阅的热门 observable。这是一个库，我想让用户能够调用某些方法并避免调用 subscribe() 来触发 observables。最初我有这个：

const q = new Queue();

q.add('foo bar baz').subscribe();   // <<< need to call subscribe


Queue.prototype.add = Queue.prototype.enqueue = function (lines) {

    lines = _.flattenDeep([lines]);

    var lockAcquired = false;

    return this.init()
        .flatMap(() => {
            return acquireLock(this)
        })
        .flatMap(() => {
            lockAcquired = true;
            return appendFile(this, lines)
        })
        .flatMap(() => releaseLock(this))
        .catch(err => {
            if (lockAcquired) {
                return releaseLock(this);
            }
            else {
                return makeGenericObservable();
            }
        })


};

为了让 observable 变热，我想我可以这样做：

const q = new Queue();

q.add('foo bar baz');  // <<< don't call subscribe

Queue.prototype.add = Queue.prototype.enqueue = function (lines) {

    lines = _.flattenDeep([lines]);

    var lockAcquired = false;

    return this.init()
        .flatMap(() => {
            return acquireLock(this)
        })
        .flatMap(() => {
            lockAcquired = true;
            return appendFile(this, lines)
        })
        .flatMap(() => releaseLock(this))
        .catch(err => {
            if (lockAcquired) {
                return releaseLock(this);
            }
            else {
                return makeGenericObservable();
            }
        })
        .publish()
        .share()  // this too?

};

但是问题是当我调用publish() 时，什么都没有发生，并且 add 方法似乎永远不会被完全调用（我假设序列中的第一个 observable 根本不会触发，因为没有调用有效的 subscribe）。但我虽然publish() 会自动调用可观察链？

如何让 add 方法返回的 observable，hot？

Answer 1

你误会了。 Hot observable确实需要像冷订阅一样订阅。不同之处在于 hot 使用一些外部生产者（如圆顶元素）并开始在订阅时收听它。另一方面，冷可观察在订阅时在内部创建生产者。

这导致您可能会错过一些带有 hot observable 的事件，因为外部生产者对订阅一无所知并且独立发出。使用冷可观察您不会错过任何东西，因为生产者是在订阅时创建的。

长话短说，您可以在热或冷可观察对象之上构建任何可观察对象链，但除非您订阅它，否则不会发生任何事情。

PS。无需将publish 与share 一起使用，因为后者是alias for .publish().refCount()。

Answer 2

如果您不想使用subscribe，您可以在您的方法中手动.connect() 或subscribe 它：

const q = new Queue();
q.add('foo bar baz');

Queue.prototype.add = Queue.prototype.enqueue = function (lines) {
    lines = _.flattenDeep([lines]);
    var lockAcquired = false;

    let add$ = this.init()
        .flatMap(() => {
            return acquireLock(this)
        })
        .flatMap(() => {
            lockAcquired = true;
            return appendFile(this, lines)
        })
        .flatMap(() => releaseLock(this))
        .catch(err => {
            if (lockAcquired) {
                return releaseLock(this);
            }
            else {
                return makeGenericObservable();
            }
        })
        .publish();

     add$.connect();
     return add$;  // optional, depends if you even need the stream outside of the add-method
};

或者作为使用内部subscribe的替代方案：

const q = new Queue();
q.add('foo bar baz');  // <<< don't call subscribe

Queue.prototype.add = Queue.prototype.enqueue = function (lines) {
    lines = _.flattenDeep([lines]);
    var lockAcquired = false;
    let add$ = this.init()
        ...
        .share();

     add$.subscribe();
     return add$;
};