使用 Wait 的方法对响应式扩展进行单元测试

Question

我正在尝试对在 IObservable 上使用 Wait() 方法的方法进行单元测试，但是我的测试永远不会完成 - Wait 永远不会完成。我的测试包含以下内容：

var scheduler = new TestScheduler();
var input1 = scheduler.CreateColdObservable<List<string>>(
   new Recorded<Notification<List<string>>>(100, Notification.CreateOnNext(new List<string> { "John", "Harry" })),
   new Recorded<Notification<List<string>>>(200, Notification.CreateOnCompleted<List<string>>())
          );

我正在使用 Moq 通过返回 input1 来设置我的方法的响应。例如

myObj.Setup(f => f.GetStrings()).Returns(input1);

myObj 的细节实际上并不重要。我启动调度程序并调用包含等待的方法（例如，在我调用的方法中某处

var results = myObj.GetStrings().Wait();

但这永远不会回来。我怀疑我使用了错误的调度程序，但我不确定。

问候 艾伦

Answer 1

总结

问题是您正在创建一个冷可观察对象并在您订阅它之前 推进调度程序。

详情

如果您在单线程测试中调用阻塞 Wait() 操作，那么您此时已经死在水中了。这是因为TestScheduler 的内部时钟仅在您调用Start() 或AdvanceXXX() 方法之一时才会提前，并且由于您有一个冷可观察对象，您指定的事件时间是相对于订阅点的。调用Start() 也有一些细微差别，我将在下面解释。

所以，由于Wait 会阻塞，您可能会尝试在另一个线程上调用它，但这仍然很棘手。考虑以下代码，它与您的类似：

void Main()
{
    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));

    // (A)

    int result = 0;

    var resultTask = Task.Run(() => { result = source.Wait(); });

    // (B)

    resultTask.Wait();

    Console.WriteLine(result);
}

此代码尝试在后台线程上等待。如果我们在点 (A) 处插入对 scheduler.Start() 的调用，那么 source.Wait() 将永远阻塞。

这是因为Start() 只会提前TestScheduler 的内部时钟，直到所有当前计划的事件都被执行。使用冷可观察，事件是相对于订阅的虚拟时间安排的。由于在 (A) 点没有订阅者，您会发现 TestScheduler.Now.Ticks 即使在调用 Start() 之后也会报告 0。

嗯。如果我们将调用 scheduler.Start() 转移到 B 点，事情会变得更糟。现在我们有一个竞争条件！这是一种竞争条件，几乎总是会导致测试在调用resultTask.Wait() 时挂起。这是因为在 scheduler.Start() 调用执行之前，resultTask 可能没有时间执行它的操作并订阅 source - 因此时间不会再提前。

因此很难实现确定性执行 - 没有很好的方法来宣布 Wait() 调用已在提前时间之前发出，因为 Wait() 调用本身会阻塞。在调用Start() 之前插入足够长的延迟会起作用，但有点违背了使用TestScheduler 的目的：

// (B)
Task.Delay(2000).Wait();        
scheduler.AdvanceBy(200);

这个问题真正向我展示的（恕我直言）是调用Wait() 并阻塞线程几乎总是一个坏主意。寻找使用 LastAsync() 之类的方法，和/或使用延续来获取异步方法的结果。

由于复杂性，我不能推荐这种方法，但这里有一个确定性的解决方案，它使用扩展方法在订阅完成时发出信号。

void Main()
{
    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));

    // (A)      
    var waitHandle = new AutoResetEvent(false);

    int result = 0;                         
    var resultTask = Task.Run(() =>
    {
        result = source.AnnounceSubscription(waitHandle).Wait();
    });

    // (B)
    waitHandle.WaitOne();   
    scheduler.Start();

    resultTask.Wait();

    Console.WriteLine(result);
}

public static class ObservableExtensions
{
    public static IObservable<T> AnnounceSubscription<T>(
        this IObservable<T> source, AutoResetEvent are)
    {
        return Observable.Create<T>(o =>
        {
            var sub = source.Subscribe(o);
            are.Set();
            return sub;
        });
    }
}

推荐的 Rx 测试方法

TestScheduler 更惯用的用法是创建一个观察者来收集结果，然后断言它们符合预期。比如：

void Main()
{
    var scheduler = new TestScheduler();
    var source = scheduler.CreateColdObservable(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));

    var results = scheduler.CreateObserver<int>();

    // here you would append to source the Rx calls that do something interesting
    source.Subscribe(results);

    scheduler.Start();

    results.Messages.AssertEqual(
        new Recorded<Notification<int>>(100, Notification.CreateOnNext(1)),
        new Recorded<Notification<int>>(200, Notification.CreateOnCompleted<int>()));   
}

最后，如果您从ReactiveTest 派生单元测试类，您可以利用OnNext、OnCompleted 和OnError 辅助方法以更易读的方式创建Recorded<Notification<T>> 实例。