如何使用 rxjs 订阅 Promise？

Question

我有一个返回承诺的函数。
我想使用 rx.js 订阅该承诺，调用此函数会将通知推送给所有订阅者。

这是我所拥有的：

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'); }
);

//here I want to push first notification
subject.fromPromise(functionThatReturnsPromise()); 

//some code

//here I want to push second notification
subject.fromPromise(functionThatReturnsPromise()); 

因此订阅者只收到一个通知。

如何解决？

Answer 1

var subject = new Rx.Subject();

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

subject.onNext(Rx.Observable.fromPromise(functionThatReturnsPromise()));

//some code

subject.onNext(Rx.Observable.fromPromise(functionThatReturnsPromise()));

这里的主题现在是一个元流（一个可观察的可观察对象）。在订阅它之前，我们通过调用mergeAll() 来“扁平化”它。在 onNext() 上，我们将 Observables 提供给主题，这就是它成为元流的原因。