c# - 如何等待 Observable 的 Observable？

Question

我有一个 IObservable对象，我想将其异步转换为列表字典。

这是我的代码，其中 GetSource 返回一个 IObservable :

await GetSource(...)
  .GroupBy(o => o.SiteId)
  .ToDictionary(g => g.Key, g => g.ToList())

当然是错的，因为结果是IDictionary<int, IObservable<IList<T>>> , 但我需要一个 IDictionary<int, IList<T>>

本质上，我需要await每一个IObservable<IList<T>> , 但我不知道如何像我确信的那样优雅地使用 Rx.NET 做到这一点。

有什么想法吗？

Answer 1

请注意，无论您做什么，在源完成之前您永远无法完成字典。因此，一种简单的方法是像这样异步获取列表:

await GetSource(...).ToList();

然后继续生成 IList<T>就像你对 GroupBy 一样和 ToDictionary - 请注意您现在正在使用 IEnumerable<T>针对完整列表实现这些运算符。

此方法假定 GetSource() 中事件的数量和时间可观察的，并且组之间的分布使得在完成的流上计算组和字典的费用并不高。由于最终您将返回一个完整的字典，无论如何我们都可能只考虑分组和创建字典条目的成本。与长事件流的序列化交付相比，这在内存数据上的速度非常快，因此我倾向于认为这种方法在少数情况下并不完美。

如果它是禁止的，可能值得同时建立每个组和列表并且事件到达，在这种情况下你可以这样做:

await GetSource(...).GroupBy(x => x.SiteId)
                    .SelectMany(x => x.ToList())
                    .ToDictionary(x => x[0].SiteId);

请注意 ToDictionary 中列表的第一个元素keySelector 将 始终存在(否则不会具体化任何组)，因此这是安全的。它看起来确实有点奇怪，但这是我能想到的取出 key 的最简单方法。

或者作为通用函数:

async Task<IDictionary<TKey, IList<T>>> ToDictionaryOfLists<T, TKey>(
    IObservable<T> source,
    Func<T, TKey> keySelector)
{        
    return await source.GroupBy(keySelector)
                       .SelectMany(x => x.ToList())
                       .ToDictionary(x => keySelector(x[0]));           
}

假设一个类:

public class Site
{
    public int SiteId { get; set; }
}

你可以这样使用:

var result = ToDictionaryOfLists(GetSource(...), x=> x.SiteId);