如何在 RxJs 中构建一个从多个源接收数据的流？答案

【问题标题】：How to build a stream in RxJs which receives data from multiple sources?如何在 RxJs 中构建一个从多个源接收数据的流？
【发布时间】：2015-08-17 08:29:19
【问题描述】：

我正在尝试构建一个包含多个组件的应用程序，这些组件使用来自服务器的事件。该应用程序是在 Angular 中构建的，并且正在使用 rxjs-ng。

我没有找到与我的用例匹配的示例。

例如，进行货币更新，您的 UI 层有多个订阅者，并且有多个数据访问和业务逻辑提供者（都在客户端）。

我已经实现了一个客户端服务，它获取货币并返回一个 observable：

update: function (from, to) {
    var urlForCurrencies = createConverterUrl(from, to);
    var observable = Rx.Observable.fromPromise(
        $http({method: 'JSONP', url: urlForCurrencies})
    );

    return observable;
}

在 UI 组件中：

var that = this;
DataService.update(currencyFrom,currencyTo).subscribe(
    function (data) {
        that.currency = data.data.CHF_EUR.val;
    },
    function (err) {
        that.error = err.message;
    }    
);

这应该只在从 UI 层请求时工作一次。

如何发送货币数据的另一个更新或从不同的视图触发更新并仍然使用相同的流/observable？

【问题讨论】：

标签： angular rxjs observable reactive-programming

【解决方案1】：

查看有关 using subjects 的文档。

Subject 类继承了 Observable 和 Observer ，从某种意义上说，它既是观察者又是可观察对象。您可以使用主题订阅所有观察者，然后将主题订阅到后端数据源。

您可以使用 Angular 服务共享主题。这样，其他组件就有可能使用 subject.subscribe(onNext, onError, onCompleted) 对事件做出反应。可以使用 subject.onNext(...) 通过主题发送更新。

// Creating a subject
var subject = new Rx.Subject(); 

...

// Handing updates over to the subject
getCurrencyUpdate().then(function(data) {
  subject.onNext(data);
});

...

// Subscribe to the subject to react on updates
subscription = subject.subscribe(
  function (x) { console.log('onNext: ' + x); },
  function (e) { console.log('onError: ' + e.message); },
  function () { console.log('onCompleted'); });
)

这样您就可以使用主题向您的订阅者广播数据。

【讨论】：