Я изо всех сил пытаюсь понять, почему следующий тест не работает:
[Fact]
public void repro()
{
var scheduler = new TestScheduler();
var count = 0;
// this observable is a simplification of the system under test
// I've just included it directly in the test for clarity
// in reality it is NOT accessible from the test code - it is
// an implementation detail of the system under test
// but by passing in a TestScheduler to the sut, the test code
// can theoretically control the execution of the pipeline
// but per this question, that doesn't work when using FromAsync
Observable
.Return(1)
.Select(i => Observable.FromAsync(Whatever))
.Concat()
.ObserveOn(scheduler)
.Subscribe(_ => Interlocked.Increment(ref count));
Assert.Equal(0, count);
// this call initiates the observable pipeline, but does not
// wait until the entire pipeline has been executed before
// returning control to the caller
// the question is: why? Rx knows I'm instigating an async task
// as part of the pipeline (that the point of the FromAsync
// method), so why can't it still treat the pipeline atomically
// when I call Start() on the scheduler?
scheduler.Start();
// count is still zero at this point
Assert.Equal(1, count);
}
private async Task<Unit> Whatever()
{
await Task.Delay(100);
return Unit.Default;
}
То, что я пытаюсь сделать, это запустить некоторый асинхронный код (представленный выше Whatever()) всякий раз, когда наблюдаемые тики. Важно отметить, что я хочу, чтобы эти вызовы были поставлены в очередь. Что еще более важно, я хочу иметь возможность контролировать выполнение конвейера с помощью TestScheduler.
Кажется, что вызов scheduler.Start() инициирует выполнение Whatever(), но он не ждет, пока он не завершится. Если я изменяю Whatever() так, чтобы он был синхронным:
private async Task<Unit> Whatever()
{
//await Task.Delay(100);
return Unit.Default;
}
тогда тест проходит, но, конечно, это побеждает цель того, чего я пытаюсь достичь. Я мог представить себе, что существует StartAsync() метод TestScheduler, который я мог ждать, но этого не существует.
Может ли кто-нибудь сказать мне, есть ли способ инициировать выполнение реактивного конвейера и ждать его завершения, даже если он содержит асинхронные вызовы?
