Reactive Extensions SelectMany和Concat

Question

据我所知,SelectMany的行为是将每个生成的值的结果有效地合并到一个流中,因此在非确定性中排序.

我如何在C#中的RxJs中执行类似于concatAll的操作.

var obs = Observable.Range (1, 10).SelectMany (x => {
return Observable.Interval (TimeSpan.FromSeconds(10 - x)).Take (3);
}).Concat();

这实际上是我想要做的,给定一个范围,然后按照它们开始的顺序等待每一个连续.显然这是一个玩具示例,但想法就在那里.

布莱尔

Answer 1

使用Select,而不是SelectMany.Concat您想要使用的重载适用于a IObservable<IObservable<T>>,因此只需投影内部序列,不要将它们展平.

var obs = Observable.Range(1, 10)
                    .Select(x => Observable.Interval(TimeSpan.FromSeconds(10 - x)).Take(3))
                    .Concat();

请注意,每个的子记录Interval都是使用Concat; 也就是说,第一个Interval在您订阅时立即开始,但所有剩余的间隔都是在没有订阅的情况下生成和排队的.它不会Concat订阅所有内容,然后以正确的顺序重播值.