RX + TPL + Тестирование

Question

В настоящее время я борюсь с проблемой тестирования функции TPL внутри оператора `RX. Проблема также в том, что у меня нет доступа к наблюдаемому.

Вот мой тест:

[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 работает отлично.
• Но в IntegrationTests с SpecFlow я не знаю, когда работа закончена без введения хакерских свойств для синхронизации.

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, который может быть проигнорирован вашим планировщиком тестов.