RX + TPL +测试

Question

我目前正在努力处理RX语句中测试TPL函数的问题。问题在于我无法访问observable。

这里是我的测试：

[Test] 
public void TestRaceConditions() 
{ 
    // Arrange 
    TestScheduler testScheduler = new TestScheduler(); 
    List<string> source = new List<string> {"Hello"}; 
    List<string> results = new List<string>(); 

    source.ToObservable().SelectMany(v => Task.Run(() => 
    { 
     Thread.Sleep(10000); // Simulate long running operation 
     return v; 
    }).ToObservable(testScheduler)) 
    .ObserveOn(testScheduler) 
    .Subscribe(s => results.Add(s)); 

    // Act 
    testScheduler.Start(); 

    Assert.That(results.Count, Is.EqualTo(1)); 
    Assert.That(results[0], Is.EqualTo("Hello")); 
} 

我周围的一派，发现了许多文章和建议，如：

• 请勿混用TPL和RX
• 创建用于TPL的包装代码，您可以在其中模拟并返回已完成的任务。 ==>但是集成测试呢？

我也试过是创建一个自定义TaskScheduler，但后来我不得不每Task.Run与Task.Factory.StartNew取代。

所以，问题是：

• 如何克服任务或在测试中观察到的只是Schedulers，我知道，当所有的元素被处理和成品控制？

UPDATE：实证

public class ResultController : IDisposable 
{ 
    private readonly IServiceClient _serviceClient; 
    private readonly IDisposable _subscription; 

    public ResultController(IEventProvider eventProvider, 
          IServiceClient serviceClient, 
          ISchedulerProvider schedulerProvider) 
    { 
     _serviceClient = serviceClient; 
     _subscription = eventProvider.Events.SelectMany(e => LoadDataAsync(e.Id)) 
              .SubscribeOn(schedulerProvider.TaskPool) 
              .ObserveOn(schedulerProvider.Dispatcher) 
              .Subscribe(OnNewDataLoaded); 
    } 

    private void OnNewDataLoaded(Data data) 
    { 

    } 

    public async Task<Data> LoadDataAsync(int id) 
    { 
     ////return Task.Run(() => 
     ////{ 
     //// Thread.Sleep(10000); 
     //// return new Data(); 
     ////}); 

     return await _serviceClient.LoadDataByIdAsync(id); 
    } 

    public void Dispose() 
    { 
     _subscription.Dispose(); 
    } 
} 

此方法适用于在生产中没有问题，但在测试中我没有得到控制的任务。

• 嘲笑ServiceClient并返回完成Task工作正常。
• 但是在使用SpecFlow的IntegrationTests中，我不知道什么时候工作完成而没有引入用于同步的黑客属性。

Answer 1

我认为，是怎么回事是你断言工作的结果，他们完成前：

// starts a work in a different thread 
// and continues the test right after firing work up 
testScheduler.Start(); 

// results here are empty for a 10 seconds 
Assert.That(results.Count, Is.EqualTo(1)); 
Assert.That(results[0], Is.EqualTo("Hello")); 

你可以做什么这里订阅任务的延续，你要安排通过Task.Run，您还可以引入一系列任务并将该任务添加到该任务并等待它们，然后才能执行检查。

由于您的模拟工作，引入自定义调度程序可能没有帮助 - 您仍然需要等待10秒才能唤醒线程。也许另一个选择是使用Task.Delay方法，而您的测试调度程序可以忽略这个方法。