与无扩展

Question

测试观测量时如何使用Task.Run虚拟时间，我有以下功能我想测试

/// 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需要太长的时间来处理，所以我在测试中放弃了多于 的消息。

问题是。我如何创建一个尊重 虚拟时间的JobTask，并允许我测试是否可以成功删除 预期的消息。

Answer 1

关键是创建一个TestScheduler知道的tick事件流。 为此目的我创建的扩展方法

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 sourced 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); 

    } 



}