我要测试以下功能
/// Items are processed asynchronously via fn as they arrive. However
/// if an item arrives before the last asynchronous operation has
/// completed then the cancellation token passed to fn will be
/// triggered enabling the task to be canceled in a best effort
/// way.
public static IObservable<U> SelectWithCancellation<T, U>
( this IObservable<T> This
, Func<CancellationToken, T, Task<U>> fn
)
{
return This
.Select(v=>Observable.FromAsync(token=>fn(token, v)))
.Switch();
}
我希望对其进行测试,并尽我所能
有效的方法如下。首先,我创建一个长期运行的任务
可以取消
public Task<string> JobTask
( CancellationToken token
, string input
)
{
return Task.Factory.StartNew(() =>
{
if ( input == "C" || input == "E" )
{
while ( !token.IsCancellationRequested ) ;
}
return input;
}
);
}
然后我测试它真的可以工作
public class SelectWithCancelationSpec : ReactiveTest
{
TestScheduler _Scheduler = new TestScheduler();
[Fact]
public void ShouldWork()
{
var o = _Scheduler.CreateHotObservable
( OnNext(100, "A")
, OnNext(200, "B")
, OnNext(300, "C")
, OnNext(400, "D")
, OnNext(500, "E")
, OnNext(500, "F")
);
List<string> actual = new List<string>();
o
.SelectWithCancellation(JobTask)
.Subscribe(v => actual.Add(v));
var delay = 100;
_Scheduler.AdvanceTo(150);
Thread.Sleep(delay);
_Scheduler.AdvanceTo(250);
Thread.Sleep(delay);
_Scheduler.AdvanceTo(350);
Thread.Sleep(delay);
_Scheduler.AdvanceTo(450);
Thread.Sleep(delay);
_Scheduler.AdvanceTo(550);
Thread.Sleep(delay);
_Scheduler.AdvanceTo(650);
var expected = new[] { "A", "B", "D", "F" };
actual
.ShouldBeEquivalentTo(expected);
}
}
问题是我不得不将
real time
引入到测试。这是因为我模拟的JobTask在真实环境中运行
从线程池中删除线程,并且不考虑虚拟时间
测试计划程序。如果我不拖延,会发生什么
在
AdvanceTo
调用之间,我丢弃的邮件多于我期望在测试中,因为JobTask花费的时间太长。
问题是。我如何创建一个尊重
虚拟时间,让我测试是否可以成功放弃
预期的消息。
最佳答案
关键是创建TestScheduler知道的滴答事件流。
为此,我创建了一个扩展方法
public static class TestSchedulerExtensions
{
public static IObservable<Unit> CreateTickObserver(this TestScheduler s, int startTick, int endTick, int tickStep)
{
var ticks = Enumerable.Repeat(1, Int32.MaxValue)
.Select(( v, i ) => i * tickStep + startTick)
.TakeWhile(v => v <= endTick)
.Select(tick => ReactiveTest.OnNext(tick, Unit.Default));
return s.CreateColdObservable(ticks.ToArray());
}
}
然后是另一种扩展方法,可帮助在测试条件下创建任务
public static Func<CancellationToken,U,Task<T>>
AsyncSelectorFactory<T, U>
( this TestScheduler s
, int duration
, int interval
, Func<CancellationToken, U, IObservable<Unit>, Task<T>> fn
)
{
var ticker = s.CreateTickObserver(0, duration, interval);
return ( c, u ) =>
{
return fn(c, u, ticker);
};
}
TaskFactory生成可以生成任务的函数,但是
在测试计划程序的控制下通过了自动收录器。那
代码可以用于引起延迟或其他原因。
请注意,上面我们等待_Ticker来源的Observable创建延迟
在任务中。现在我们的测试用例看起来像
现在测试很简单
public class SelectWithCancelationSpec : ReactiveTest
{
TestScheduler _Scheduler = new TestScheduler();
[Fact]
public void ShouldWork()
{
var o = _Scheduler.CreateColdObservable
( OnNext(100, "A")
, OnNext(200, "B")
, OnNext(300, "C")
, OnNext(400, "D")
, OnNext(500, "E")
, OnNext(600, "F")
);
int cancelCount = 0;
var job = _Scheduler.AsyncSelectorFactory<string,string>
( 1000
, 10
, async ( token, input, ticker ) => {
if ( input == "C" || input == "E" )
{
await ticker.TakeWhile(v => !token.IsCancellationRequested);
cancelCount++;
}
return input;
});
var actual = _Scheduler.Start(() =>
{
return o.SelectWithCancellation(job);
}
, created: 0
, subscribed: 1
, disposed: 1000
);
var expected = new[] { "A", "B", "D", "F" };
cancelCount.Should().Be(2);
actual.Messages.Select(v=>v.Value.Value)
.ShouldBeEquivalentTo(expected);
}
}
关于c# - 在使用Reactive Extensions测试Observable时如何在Task.Run中使用虚拟时间,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/17852061/