.net - .NET Rx Koans: このテスト ケースが失敗するのはなぜですか?

Answer 1

簡潔な答え：

結果を確認する前に、テストの非同期部分を実行するのに十分な時間がありません。

より長い答え：

このテストが実行する一連のアクションは、次のようなものです。

• IObservableデリゲートを非同期的に呼び出す を設定する
• それを、これらの非同期呼び出しのうちの 2 つのマージ結果である別のものにチェーンしIObservable、合計します
• Subscribe結果IObservableの に、メソッドの 2 つの非同期呼び出しを引き起こします
• どっ！デリゲートにがあるThread.Sleepため、非同期呼び出しがブロックされます。
• すぐに結果を確認します。結果はもちろん 0 になります。ブロックされた 2 つの非同期呼び出しは「終了」していません。

これを「修正」する方法はいくつかあります。

• を削除しますThread.Sleep
• を変更して呼び出しを同期に変更しますがBeginInvoke、これにはテスト全体の再構築が必要になります
• HistoricalSchedulerの代わりにa を使用Immediateする

HistoricalSchedulerRx の単体テストを行う場合は、a を使用することを強くお勧めします。基本的には、仮想時間を前後にジャンプできます。これは、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);

Answer 2

これが同期バージョンの外観です - あなたが持っているものの最も直接的なバージョンです。Single()オブザーバブルが完了するまでブロックします。通常、ブロックは避けたいものですが、いじるだけなら問題ありません。

public void AsynchronousRunInParallel()
{
    Func<int, int> inc = (int x) =>
    {
        Thread.Sleep(1500);
        return x + 1;
    };

    var incAsync = Observable.FromAsyncPattern<int, int>(inc.BeginInvoke,
                                                         inc.EndInvoke);

    int sum = incAsync(1).Merge(incAsync(9)).Sum().Single();

    Assert.AreEqual(12, sum);
}

そして、以下を使用した非同期 TPL バージョンawait:

public async Task AsynchronousRunInParallel()
{
    Func<int, int> inc = (int x) =>
    {
        Thread.Sleep(1500);
        return x + 1;
    };

    var incAsync = Observable.FromAsyncPattern<int, int>(inc.BeginInvoke,
                                                         inc.EndInvoke);

    int sum = await incAsync(1).Merge(incAsync(9)).Sum();

    Assert.AreEqual(12, sum);
}

最後に、Rx を使用する非同期のものDo()です。たとえば、これがより大きな操作の一部であるとします。

public async Task AsynchronousRunInParallel()
{
    Func<int, int> inc = (int x) =>
    {
        Thread.Sleep(1500);
        return x + 1;
    };

    var incAsync = Observable.FromAsyncPattern<int, int>(inc.BeginInvoke,
                                                         inc.EndInvoke);

    await incAsync(1).Merge(incAsync(9)).Sum().Do(sum =>
    {
        Assert.AreEqual(12, sum);
    });
}