Rx扩展以保持N秒内生成事件

Question

什么是正确的Rx扩展方法(在。net中)以保持N秒生成事件?

通过'keep generating events for N seconds'，我的意思是它将在一个循环中继续生成事件，因为DateTime。现在到DateTime。Now + TimeSpan.FromSeconds(N)

我正在研究一种遗传算法，它将产生许多假设，并将最成功的假设传播给下一代。需要以某种优雅的方式约束这个家伙

补充道:

我实际上已经意识到我需要做拉而不是推，并想出了这样的东西:

public static class IEnumerableExtensions
{
    public static IEnumerable<T> Pull<T>(this IEnumerable<T> enumerable, int? times = null)
    {
        if (times == null) 
            return enumerable.ToArray();
        else
            return enumerable.Take(times.Value).ToArray();
    }
    public static IEnumerable<T> Pull<T>(this IEnumerable<T> enumerable, TimeSpan timeout, int? times = null)
    {
        var start = DateTime.Now;
        if (times != null) enumerable = enumerable.Take(times.Value);
        using (var iterator = enumerable.GetEnumerator())
        {
            while (DateTime.Now < start + timeout && iterator.MoveNext())
                yield return iterator.Current;
        }
    }
}

的用法是:

var results = lazySource.SelectMany(item =>
{
    //processing goes here
}).Pull(timeout: TimeSpan.FromSeconds(5), times: numberOfIterations);

Answer 1

可能有更简洁的方法，但是您可以使用:

// This will generate events repeatedly
var interval = Observable.Interval(...);
// This will generate one event in N seconds
var timer = Observable.Timer(TimeSpan.FromSeconds(N));
// This will combine the two, so that the interval stops when the timer
// fires
var joined = interval.TakeUntil(timer);

我已经很长时间没有做任何Rx了，所以如果这是不正确的，我很抱歉-但值得一试…

Answer 2

Jon的帖子非常正确，但是我注意到你的编辑建议你创建自己的扩展方法来做到这一点。我认为如果你只使用内置的操作符会更好。

//LinqPad sample
void Main()
{
    var interval = Observable.Interval(TimeSpan.FromMilliseconds(250));  
    var maxTime = Observable.Timer(TimeSpan.FromSeconds(10));
    IEnumerable<int> lazySource = Enumerable.Range(0, 100);
    lazySource.ToObservable()
            .Zip(interval, (val, tick)=>val)
            .TakeUntil(maxTime)
            .Dump();
}

* ie。易于其他开发人员维护和理解