从 IObservable 转换为 IEnumerable 的反应式扩展答案

【问题标题】：Reactive Extensions converting from IObservable to IEnumerable从 IObservable 转换为 IEnumerable 的反应式扩展
【发布时间】：2017-04-27 15:20:52
【问题描述】：

给定一个 Observable，它会在调用 Subscribe 后开始推送事件。如何将序列转换为 IEnumerable。

2010 年 Jon Skeet 写了关于他对 RX 的第一印象并写了这篇文章，这将相当于我所需要的。

public static List<T> ToList<T>(this IObservable<T> source) 
{ 
    List<T> ret = new List<T>(); 
    source.Subscribe(x => ret.Add(x)); 
    return ret; 
}

https://codeblog.jonskeet.uk/2010/01/16/first-encounters-with-reactive-extensions/

虽然此解决方案可行，但似乎应该有更好的方法。

我们真正感兴趣的是能够以类似于订阅的方式await observable。

await observable.ToList()

但是，这似乎对我不起作用。

我也研究过这两种方法

observable.ToEnumerable
observable.Next

更多信息

我打算用它来写一些单元测试。

我正在使用 ReplaySubject 并使用测试替代品执行一系列操作。
通过OnNext 记录对替代者的呼叫。
我可以获得调用列表并断言它们是正确的。

【问题讨论】：

Jon Skeet 的文章写于 2010 年，当时 Rx 只是一个研究项目，async/await 是未来两年。 By 2011 Observable 有自己的ToList() 扩展方法
@PanagiotisKanavos 这样您就可以将 IObservable 更改为 IObservable> IObservable<IList<TSource>> ToList<TSource>(this IObservable<TSource> source) MSDN Daniel 正在询问 IObservable 到 IEnumerable 的转换
@lerthe61 检查 Shlomo 的答案，该答案显示了如何做到这一点。 ToEnumerable() 提供了一个阻塞选项。 Jon Skeet 的文章是在 Rx 是一个研究项目时写回来的，不再适用

标签： c# .net system.reactive

【解决方案1】：

有趣的是，斯基特先生的实现非常糟糕。它不会阻塞，因此仅当源恰好是重播主题（或类似主题）时才会将值泵入列表。所以对于大多数 Observable 序列，这只会返回一个空列表。我确信 Jon 知道这一点，但可能在翻译中迷路了。

ToList() 是您想要的，所以我假设您可能没有完成？如果您只想要已知数量的元素，那么Take(x).ToList() 就足够了。

【讨论】：

我强烈建议编辑这个答案。 Jon Skeet 的文章写于 2010 年，当时还没有 async/await。 Reactive Extensions 那时只是一个研究项目。没有ToList(),

【解决方案2】：

编辑：

如果您无法完成如下所述的流，则使用await 将无休止地等待。我会以其他方式进行测试。大家可以看看 Nuget Package Microsoft.Reactive.Testing，里面包含了一些强大的测试工具。或者只是使用Subscribe 并在那里断言。

await 等待 observable 完成或出错。您可能只需要向您的ReactiveSubject 添加一个OnCompleted 调用就可以了。以下代码适用于我：

var rs = new ReplaySubject<int>();

rs.OnNext(1);
rs.OnNext(2);
rs.OnNext(3);
rs.OnCompleted();
var l = await rs.ToList();

//l is equivalent to  new List<int>() {1, 2, 3};

【讨论】：

我刚刚意识到我无法完成直播
更新答案。
在订阅方法中似乎无法断言
我也不需要列表，如果有帮助，我只需要前 n 个事件，所以我可以说 take(3).toList()