如何在不引入竞争条件的情况下等待RX主题的响应？

Question

我有一个服务,允许调用者发送命令并异步接收响应.在实际应用程序中,这些操作相当断开(某些操作将发送命令,响应将独立处理).

但是,在我的测试中,我需要能够发送命令,然后在继续测试之前等待(第一个)响应.

响应是使用RX发布的,我对代码的第一次尝试是这样的:

service.SendCommand("BLAH");
await service.Responses.FirstAsync();

这个问题是,FirstAsync只有在await已经被命中之后响应到达时才会起作用.如果服务进程很快,那么测试将挂起await.

我下次尝试解决此问题的方法是FirstAsync()在发送命令之前调用,以便即使它在等待之前到达它也会产生结果:

var firstResponse = service.Responses.FirstAsync();
service.SendCommand("BLAH");
await firstResponse;

但是,这仍然以同样的方式失败.似乎它只是在它被开始聆听时await被击中(GetAwaiter被称为); 所以存在完全相同的竞争条件.

如果我ReplaySubject用缓冲区(或计时器)将我的主题更改为a ,那么我可以"解决"这个问题; 但是在我的生产课程中这样做是没有意义的; 它只会用于测试.

在RX中能够做到这一点的"正确"方法是什么？如何以不会引入竞争条件的方式设置将在流上接收第一个事件的内容？

这是一个以"单线程"方式说明问题的小测试.此测试将无限期挂起:

[Fact]
public async Task MyTest()
{
    var x = new Subject<bool>();

    // Subscribe to the first bool (but don't await it yet)
    var firstBool = x.FirstAsync();

    // Send the first bool
    x.OnNext(true);

    // Await the task that receives the first bool
    var b = await firstBool; // <-- hangs here; presumably because firstBool didn't start monitoring until GetAwaiter was called?


    Assert.Equal(true, b);
}

我甚至尝试在我的测试中调用Replay(),认为它可以缓冲结果; 但这并没有改变任何事情:

[Fact]
public async Task MyTest()
{
    var x = new Subject<bool>();

    var firstBool = x.Replay();

    // Send the first bool
    x.OnNext(true);

    // Await the task that receives the first bool
    var b = await firstBool.FirstAsync(); // <-- Still hangs here


    Assert.Equal(true, b);
}

Answer 1

你可以这样做 AsyncSubject

[Fact]
public async Task MyTest()
{
    var x = new Subject<bool>();

    var firstBool = x.FirstAsync().PublishLast(); // PublishLast wraps an AsyncSubject
    firstBool.Connect();

    // Send the first bool
    x.OnNext(true);

    // Await the task that receives the first bool
    var b = await firstBool;


    Assert.Equal(true, b);
}

AsyncSubject在OnComplete调用之前基本缓存最后收到的值,然后重放它.

Answer 2

好问题丹尼.这给很多Rx新手带来了麻烦.

FlagBug上面有一个可接受的答案,但是刚刚添加单行可能更容易

var firstBool = x.Replay();
firstBool.Connect();   //Add this line, else your IConnectableObservable will never connect!

这种测试方式还可以.但还有另一种方式,根据我的经验,一旦他们使用Rx一段时间后,人们就会迁移到这里.我建议你直接去看这个版本!但是让我们慢慢来......

(请原谅切换回NUnit,因为我在这台PC上没有xUnit跑步者)

在这里,我们只需将值添加到a List<T>中即可生成.然后我们可以在断言中检查列表的内容:

[Test]
public void MyTest_with_List()
{
    var messages = new List<bool>();
    var x = new Subject<bool>();

    x.Subscribe(messages.Add);

    // Send the first bool
    x.OnNext(true);

    Assert.AreEqual(true, messages.Single());
}

对于这些超级简单的测试,没关系,但我们错过了序列终止的一些保真度,即它是完成还是错误？

我们可以通过使用Rx(Rx-testing Nuget)的测试工具进一步扩展这种测试方式.在这个测试中,我们使用MockObserver/ ITestableObserver<T>我们(恼人地)从一个TestScheduler实例获取.注意我已经使测试夹具/类扩展ReactiveTest

[TestCase(true)]
[TestCase(false)]
public void MyTest_with_TestObservers(bool expected)
{
    var observer = new TestScheduler().CreateObserver<bool>();
    var x = new Subject<bool>();

    x.Subscribe(observer);

    x.OnNext(expected);

    observer.Messages.AssertEqual(
        OnNext(0, expected));
}

对于创建测试调度程序的需求,以及指定我们看到消息的预期时间,这似乎是一个小改进,甚至可以说是倒退.但是,只要您开始引入更复杂的Rx测试,这就变得非常有价值.

您可以进一步扩展测试,甚至可以预先生成源序列,并指定何时在虚拟时间内播放值.在这里我们放弃主题的用法并指定在1000ticks中我们将发布一个值(expected).在断言中,我们再次检查值以及接收值的时间.由于我们现在正在介绍虚拟时间,我们还需要说明何时需要时间推进.我们通过电话来做到这一点testScheduler.Start();

[TestCase(true)]
[TestCase(false)]
public void MyTest_with_TestObservables(bool expected)
{
    var testScheduler = new TestScheduler();
    var observer = testScheduler.CreateObserver<bool>();
    var source = testScheduler.CreateColdObservable(
        OnNext(1000, expected));

    source.Subscribe(observer);
    testScheduler.Start();

    observer.Messages.AssertEqual(
        OnNext(1000, expected));
}

我在这里写了更多关于测试Rx的文章