将观察者存储在全局变量中并稍后使用答案

【问题标题】：Storing observer in a global variable and use it later将观察者存储在全局变量中并稍后使用
【发布时间】：2017-07-31 18:41:18
【问题描述】：

我如何创建一个 observable 并稍后在代码中生成下一个值？由于其他异步事件，我希望能够从代码的不同部分调用 onNext。

这是我尝试过的，但不起作用：

var Rx = require('rx');

var GlobalObserver;

var source = Rx.Observable.create(observer => {
    GlobalObserver = observer;
});

var subscription = source.subscribe(
    x => console.log('onNext: %s', x),
    e => console.log('onError: %s', e),
    () => console.log('onCompleted')
);

//...later in the code, as a result of another async event:


    GlobalObserver.onNext(someData);
    ...
    GlobalObserver.onNext(someOtherData);

【问题讨论】：

标签： javascript node.js rxjs

【解决方案1】：

你需要的是某种主题。 ReplaySubject、BehaviorSubject、Subject等

创建一个主题，然后您可以通过subject.subscribe(...) 订阅它。您也可以通过subject.onNext(...) 添加到流中。

例如：

var subject = new Rx.Subject();

var subscription = subject.subscribe(
    function (x) { console.log('onNext: ' + x); },
    function (e) { console.log('onError: ' + e.message); },
    function () { console.log('onCompleted'); }
);

subject.onNext(1);
// => onNext: 1

subject.onNext(2);
// => onNext: 2

subject.onCompleted();
// => onCompleted

subscription.dispose();

一个更具体的用例（每次成功的 HTTP 响应返回时都会添加到可观察流中）：

var httpResponseStream = new Rx.Subject();

var subscription = httpResponseStream.subscribe(function (response) { 
    console.log('HTTP response success: ', response); 
});

makeAJAXCall().then(function (response) {
    httpResponseStream.onNext(response);
});

正如另一位用户所说，如果您使用 V5，请确保将所有 onNext 更改为 next。如果您使用的是 V4，请坚持onNext。

【讨论】：

【解决方案2】：

我假设这是因为您使用的是 rxjs 版本 ^5.0.0 并阅读了版本 ^4.0.0 的文档。

对于 RxJs 版本 ^5.0.0 它应该是 observer.next(value) 而不是 observer.onNext(value)。

在这里您可以找到 RxJs 版本的docs ^5.0.0

【讨论】：

这绝对应该是评论，而不是答案。