c# - 如何在不引入竞争条件的情况下等待 RX Subject 的响应？

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，那么我可以“解决”这个问题；然而，在我的生产类(class)中这样做是没有意义的；它只会用于测试。

能够在 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 之前接收到的最后一个值，然后重播它。