javascript - 使用RxJS合并未知数量的可观察对象

我目前正在尝试创建一个小项目，以演示使用RxJS进行响应式编程。目的是向我的同事们展示这个东西在那里，值得研究。我对框架没有经验，因此使事情变得复杂。
我正在尝试扩展我的另一个演示以利用RxJS。
这不是一个非常复杂的演示，基本上我可以添加任意数量的小表格，这将导致通过一个小公式计算出一个数字，并且有一个按钮可以汇总所有表格的值。
让公式在表格内计算很容易，但是我认为我可以走得更远。
我想通过合并的Observable自动完成求和运算。
我发现的唯一解决方案是遵循以下原则:

//dummy observables
var s1 = Rx.Observable.Interval(100);
var s2 = Rx.Observable.Interval(200);

//Combine the observables
var m = s1.combineLatest(s2, function(x,y){return x+y});

//Subscribe to the combined observable
var sub = m.subscribe(function(x){console.log(x)});
//A new observable is created
var s3 = Rx.Observable.Interval(300);
//Now I need to update all my subscriptions, wich is a pain.
m = m.combine(s3, function(x,y){return x+y});
sub.dispose();
sub=m.subscribe(function(x){console.log(x)});

我想我可以得到另一个观察到的消息，通知我的订阅进行自我更新-因为知道所有订阅者的工作方式将使整个体系结构变得无用，但这听起来像是对这样一个任务的过大杀伤力，而我不仅仅是说演示，我真的无法想象有一个“每天”的现实世界示例，在这种示例中，像这样的体系结构将使事情变得更加清洁，而不是仅关注任何更改并从表单中“主动”获取计算值。
我可能会主动进行处理，并在处理表单的模块内部进行值的求和，并使外部用户可以观察到“m”，从而将我的值从模块内部插入其中。
这是正确的方法吗？我想是的，因为它们归我的模块所有，所以我应该完全控制它们所发生的事情，但是我对那些经验丰富的人对此有什么看法非常感兴趣。

最佳答案

我认为您不会找到可以直接满足您需求的运算符(operator)。

但是，精心设计自己的运算符没有任何问题:

var source = //An observable of observables of form data

Observable.prototype.combineLatestObservable = function(resultSelector) {
  var source = this;
  return Rx.Observable.create(function(obs) {
    var disposable = new Rx.SerialDisposable();
    var sources= [];
    return source.subscribe(
      function(x) {
        //Update the set of observables
        sources.push(x);
        //This will dispose of the previous subscription first
        //then subscribe to the new set.
        disposable.seDisposable(Rx.Observable.combineLatest(sources, resultSelector)
                                             .subscribe(obs));
      },
      function(e) { obs.onError(e); },
      function() { obs.onCompleted(); });
  }).share();
}

或者，如果您想与运营商合作:

//Have to use arguments since we don't know how many values we will have
function sums() {
  var sum = 0;
  for (var i = 0, len = arguments.length; i < len; ++i) {
    sum += arguments[i];
  }
  return sum;
}

source
//Capture the latest set of Observables
.scan([], function(acc, x) {
  acc.push(x);
  return acc;
})
//Dispose of the previous set and subscribe to the new set
.flatMapLatest(function(arr) {
  return Observable.combineLatest(arr, sums);
})
//Don't know how many subscribers you have but probably want to keep from
//recreating this stream for each
.share();

关于javascript - 使用RxJS合并未知数量的可观察对象，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/32006174/