如何组合来自分组的聚合

Question

我有一个尾部实现，它将文本文件中的新行推送到Subject。该文件有以下数据：

source1, 1
source2, 3
source1, 2
source1, 1
source3, 10

我正在尝试按来源创建最新聚合的视图，比如说运行总和。

第一行之后：

source1, 1

第二行之后：

source1, 1
source2, 3

第三行之后：

source1, 3
source2, 3

第四行之后：

source1, 4
source2, 3

最后一行之后：

第四行之后：

source1, 4
source2, 3
source3, 10

这是我目前所得到的（在 LinqPad 中）：

var source = Observable.Generate<int,Measurment>(0,
    current => current <= 10,
    current => current + 1,
    current => current % 3 == 0
        ? new Measurment { Source = "Source1", Value = current }
        : current != 10
            ? new Measurment { Source = "Source2", Value = current }
            : new Measurment { Source = "Source3", Value = current }
);
var grouped = source
.GroupBy(m => m.Source)
.Select(g =>g.Scan((acc,current) =>
    new Measurment { Source = acc.Source, Value = acc.Value + current.Value }));

grouped.Dump();

}
struct Measurment
{
    public string Source;
    public int Value;

我收到IObservable<IObservable<Measurment>>。内部 IObservable 具有正确聚合的值。现在我需要合并这些流并推出一个关于任何 Observables 变化的测量列表，有什么建议吗？

Answer 1

与this answer 和this one 类似，CombineLatest 的Rxx 重载将IObservable<IObservable<T>> 转换为IObservable<IList<T>>，这应该可以满足您的需求。我认为这个函数真的应该在主 Rx 库中，因为它非常有用。