将 IObservable<IObservable<T>> 组合成 IList<T>答案

【问题标题】：Combining IObservable<IObservable<T>> into IList<T>将 IObservable<IObservable<T>> 组合成 IList<T>
【发布时间】：2013-04-18 22:57:20
【问题描述】：

我的传入连接显示为：

IObservable<double>

每次建立新连接时，我都会通过以下方式通知它：

IObservable<IObservable<double>> m_IncomingConnections;

我的问题是我想构建一个我可以用来做的操作员

IObservable<IList<double>> m_joined = m_IncomingConnections.JoinAll();

因此假设一开始我有三个传入连接，m_joined observable 将推送具有三个元素的列表。

|--a1----a2----------a3-------
|b1---b2----b3--b4---------b5-
|---c1----c2---c3--c4---------

结果：

>[a1,b1,c1]
>[a1,b2,c1]
>[a2,b2,c1]
>[a2,b2,c2]
>[a2,b3,c2]

你明白了，但是如果有新的连接进来，我希望数组在新连接推送新值后立即变为 4 大小。

所有这些都是为了让我现在可以使用

m_joined.Select(vv => vv.Average());

每次推入新值时，对所有流的平均值进行流式传输。

关于为 Rx 实现这样的扩展有什么建议吗？

感谢您的帮助。

【问题讨论】：

标签： .net c#-4.0 system.reactive

【解决方案1】：

主要问题是将可观察对象列表转换为可观察对象列表。

IObservable<List<T>> CombineNLatest<T>(List<IObservable<T>> list){

    return list
        .Aggregate
            ( Observable.Return(new List<T>())
            , (a, n) => Observable.CombineLatest
                ( a
                , b
                , (l, e) => l.Concat(e).ToList()));
}

类似的东西虽然我没有测试过。现在您遇到了需要解决的问题建立你的清单>

IObservable<List<IObservable<T>> 
CollectList<T>
(this IObservable<IObservable<T>> input)
{
    var initial = new List<IObservable<T>>();
    return Observable.Create(observer => {

        return input.Subscribe( o => {

            initial.Append(o);
            observer.OnNext(initial);

        }

    });
}

再次未经测试，但我希望总体思路很清楚。现在将它们链接在一起

IObservable<IObservable<T>> input = ...;
IObservable<IObservable<List<T>>> intermediate = input
   .CollectList()
   .Select( list => CombineNLatest(list) );

IObservable<List<T>> final = intermediate.Switch();

【讨论】：