使用 .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 中进行了测试。例如，只关心枚举器实例在结果可观察量被释放后会发生什么。任何更正表示赞赏。

Answer 3

你需要一些东西来安排通知观察者，每个值都取自枚举。您可以在 Rx 调度程序上使用递归调度重载。

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

在每次定时调用时，只需调用 enumerator.MoveNext() 并调用 OnNext(enumerator.Current) ，最后OnCompleted MoveNext() 返回 false 时。这几乎是最基本的方式。

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

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

Answer 4

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

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

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

Answer 5

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

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

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 代码是什么。