c# – 无效扩展固定当呼叫长于间隔长度时异步呼叫之间的间隔

这是我的Interval定义：

m_interval = Observable.Interval(TimeSpan.FromSeconds(5),m_schedulerProvider.EventLoop)
                .ObserveOn(m_schedulerProvider.EventLoop)
                .Select(l => Observable.FromAsync(DoWork))
                .Concat()
                .Subscribe();

在上面的代码中,我在Interval和amp;中提供了IScheduler.从SchedulerProvider观察,以便我可以更快地进行单元测试(TestScheduler.AdvanceBy).此外,DoWork是一种异步方法.

在我的特定情况下,我希望每5秒调用一次DoWork函数.这里的问题是我希望5秒是DoWork结束和另一个开始之间的时间.因此,如果DoWork执行时间超过5秒,那么说10秒,第一个呼叫将是5秒,第二个呼叫是15秒.

不幸的是,以下测试证明它不像那样：

[Fact]
public void MultiPluginStatusHelperShouldWaitForNextQuery()
{    
    m_queryHelperMock
        .Setup(x => x.CustomQueryAsync())
        .Callback(() => Thread.Sleep(10000))
        .Returns(Task.Fromresult(new QueryCompletedEventData()))
        .Verifiable()
    ;

    var multiPluginStatusHelper = m_container.GetInstance<IMultiPluginStatusHelper>();
    multiPluginStatusHelper.MillisecondsInterval = 5000;
    m_testSchedulerProvider.EventLoopScheduler.AdvanceBy(TimeSpan.FromMilliseconds(5000).Ticks);
    m_testSchedulerProvider.EventLoopScheduler.AdvanceBy(TimeSpan.FromMilliseconds(5000).Ticks);

    m_queryHelperMock.Verify(x => x.CustomQueryAsync(),Times.Once);
}

DoWork调用CustomQueryAsync,测试失败,表示已调用两次.它应该只被调用一次,因为.Callback(()=> Thread.Sleep(1000))强制延迟.

我在这做错了什么？

我的实际实施来自this example.

解决方法

通常在轮询一些不可观察的数据源时会出现这个问题.当我遇到它时,我使用了一段时间后写的RepeatAfterDelay运算符：

public static IObservable<T> RepeatAfterDelay<T>(this IObservable<T> source,TimeSpan delay,IScheduler scheduler)
{
    var repeatSignal = Observable
        .Empty<T>()
        .Delay(delay,scheduler);

    // when source finishes,wait for the specified
    // delay,then repeat.
    return source.Concat(repeatSignal).Repeat();
}

这就是我使用它的方式：

// do first set of work immediately,and then every 5 seconds do it again
m_interval = Observable
    .FromAsync(DoWork)
    .RepeatAfterDelay(TimeSpan.FromSeconds(5),scheduler)
    .Subscribe();

// wait 5 seconds,then do first set of work,then again every 5 seconds
m_interval = Observable
    .Timer(TimeSpan.FromSeconds(5),scheduler)
    .SelectMany(_ => Observable
        .FromAsync(DoWork)
        .RepeatAfterDelay(TimeSpan.FromSeconds(5),scheduler))
    .Subscribe();