如何在异步场景中测试可观察的 Rx？

Question

我已将问题分解为一个示例。 处理器执行异步操作并通过 Rx observable 发布结果。

public class Tests
{
    [Test]
    public async Task Receiving_an_async_result_should_succeed()
    {
        var p = new Processor();

        p.Process("message");

        var res = await p.Results;

        res.Should().NotBeNullOrEmpty();
    }
}

public class Processor
{
    private readonly ReplaySubject<string> _subject = new ReplaySubject<string>();

    public async void Process(string message)
    {
        try
        {
            // Fake some async processing here!
            var r = await Task.Factory.StartNew(() => message.Reverse().ToString());

            _subject.OnNext(r);
        }
        catch (Exception e)
        {
            _subject.OnError(e);
        }
    }

    public IObservable<string> Results { get { return _subject; } }
}

这里的问题是我不知道如何正确设置测试用例。看来期待的结果永远不会到来！

Answer 1

结果是一个IObservable<T> - 因此在它发送一个OnCompleted() 之前它还没有完成。这就是您的 await 没有返回的原因。

我怀疑您只想等待第一个结果（看起来您打算反复致电Process）-因此，如果您更改此行：

var res = await p.Results;

到这里：

var res = await p.Results.Take(1);

你会有更多的运气。