system.reactive - 使用 .NET 响应式扩展定期调度 IEnumerable

Question

比如说我有一个可枚举的

dim e = Enumerable.Range(0, 1024)

我希望能够做到

dim o = e.ToObservable(Timespan.FromSeconds(1))

这样可观察对象每秒都会生成值，直到可枚举对象用完为止。我想不出一个简单的方法来做到这一点。

Answer 1

您可以将 Interval 与 Zip 一起使用来获得所需的功能：

var sequence = Observable.Interval(TimeSpan.FromSeconds(1))
                         .Zip(e.ToObservable(), (tick, index) => index)

Answer 2

我也在寻找解决方案，在阅读了rx 的介绍之后，我自己做了一个：有一个Observable.Generate()重载，我用它来制作我自己的ToObservable()扩展方法，以TimeSpan句号为周期：

public static class MyEx {
    public static IObservable<T> ToObservable<T>(this IEnumerable<T> enumerable, TimeSpan period) 
    {
        return Observable.Generate(
            enumerable.GetEnumerator(), 
            x => x.MoveNext(),
            x => x, 
            x => x.Current, 
            x => period);
    }
    public static IObservable<T> ToObservable<T>(this IEnumerable<T> enumerable, Func<T,TimeSpan> getPeriod) 
    {
        return Observable.Generate(
            enumerable.GetEnumerator(), 
            x => x.MoveNext(),
            x => x, 
            x => x.Current, 
            x => getPeriod(x.Current));
    }
}

已经在LINQPad中测试过。只关心在结果 observable 被处理之后枚举器实例会发生什么。任何更正表示赞赏。

Answer 3

你需要一些东西来安排通知观察者从 Enumerable 中获取的每个值。您可以在 Rx 调度程序上使用递归调度重载。

Public Shared Function Schedule ( _
    scheduler As IScheduler, _
    dueTime As TimeSpan, _
    action As Action(Of Action(Of TimeSpan)) _
) As IDisposable

在每个预定的调用中，只需 callenumerator.MoveNext()和 call OnNext(enumerator.Current)，最后OnCompletedwhenMoveNext()返回 false。这几乎是最简单的方法。

另一种表达您的要求是将其重述为“对于一个序列，每个值之间有一个最小间隔”。

看到这个答案。测试用例类似于您的原始问题。

Answer 4

你总是可以做这个非常简单的方法：

dim e = Enumerable.Range(0, 1024)
dim o = e.ToObservable().Do(Sub (x) Thread.Sleep(1000))

当您订阅o这些值时，需要一秒钟才能生成。

Answer 5

我只能假设您正在使用 Range 来简化您的问题。

您是否希望 Enumerable 推送的每个值都延迟一秒？

var e = Enumerable.Range(0, 10);
var o = Observable.Interval(TimeSpan.FromSeconds(1))
                  .Zip(e, (_,i)=>i);

还是您只希望每秒推送 Enumerable 的最后一个值。即从 Enumerable 读取，它在您枚举它时正在评估（可能是一些 IO）。在这种情况下，CombineLatest比Zip更有用。

或者您可能只想每秒获取一个值，在这种情况下，只需使用 Observable.Interval 方法

var o = Observable.Interval(TimeSpan.FromSeconds(1));

如果您解释您的问题空间，那么社区将能够更好地帮助您。

李

*请原谅 C# 的答案，但我不知道等效的 VB.NET 代码是什么。