.NET Rx Koans : Why does this test case fail?

Question

在寻找学习 Rx 的 Material 时，我发现了这个:Reactive Extensions (Rx) Koans .简介:

Definition of ‘Koan’
Kōans is a zen word meaning the enlightenment or awakening of a person, usually through a puzzle or riddle. The most common one is “What is the sound of one hand clapping?”

它由许多简短的测试用例组成，教授 Rx 的不同方面。

其中一个应该直观地通过，但是，它失败了。你能解释一下为什么吗？

这里是完整的:

    [TestMethod]
    [Timeout(___)] //"Fill in the blanks" - I tried several values, e.g. 4000. No changes.
    public void AsynchronousRunInParallel()
    {
        Func<int, int> inc = (int x) =>
                                {
                                    // I set a breakpoint here and it's never hit.

                                    Thread.Sleep(1500);
                                    return x + 1;
                                };
        double result = 0;
        var incAsync = Observable.FromAsyncPattern<int, int>(inc.BeginInvoke,
                                                             inc.EndInvoke);
        incAsync(1).Merge(incAsync(9)).Sum()
                               .SubscribeOn(Scheduler.Immediate)
                               .Subscribe(n => result = n);

        Assert.AreEqual(12, result);
                    //the failing message says: 'expected 12, got 0'
    }

Answer 1

简答:

在检查结果之前，您没有足够的时间让测试的异步部分执行。

更长的答案:

这个测试执行的 Action 序列有点像:

• 设置一个将异步调用委托(delegate)的 IObservable
• 将其链接到另一个 IObservable 中，该 IObservable 是其中两个异步调用的合并结果，加在一起
• Subscribe 到生成的 IObservable，导致方法的两次异步调用
• 哇!委托(delegate)中有一个Thread.Sleep，所以异步调用被阻塞了!
• 立即检查结果，结果当然是 0 - 两个阻塞的异步调用还没有“完成”

有很多方法可以“解决”这个问题:

• 删除 Thread.Sleep
• 通过更改 BeginInvoke 将调用更改为同步调用，尽管这需要对测试进行整体重组
• 使用 HistoricalScheduler 而不是 Immediate 一个

强烈建议在尝试对 Rx 内容进行单元测试时使用 HistoricalScheduler - 基本上，它可以让您在虚拟时间中前后跳转，这是测试 Rx 查询等与时间相关的代码的关键特性:

var theTardis = new HistoricalScheduler();

Func<int, int> inc = (int x) =>
{
    theTardis.Sleep(TimeSpan.FromMilliseconds(1500));
    return x + 1;
};
double result = 0;
var incAsync = Observable.FromAsyncPattern<int, int>(inc.BeginInvoke,inc.EndInvoke);

incAsync(1).Merge(incAsync(9)).Sum()
    .SubscribeOn(theTardis)
    .Subscribe(n => result = n);

// To the FUTURE!
theTardis.AdvanceBy(TimeSpan.FromSeconds(5));

Assert.AreEqual(12, result);