如何异步消费ServiceStack的ServerEventsClient OnMessage

Question

我正在寻找使这部分代码异步的方法。

var sse = new ServerEventsClient(apiUrl)
{
  OnMessage = HandleResponse
};

我在https://docs.servicestack.net/csharp-server-events-client#using-c-asyncawait-friendly-apis 上查看了“使用 C# Async/Await 友好 API”，但我不确定提供的代码是否仅适用于未来的一条消息或即将到来的每条消息。我将发送很多请求，我希望服务器通过 SSE 响应，所以我希望处理每个响应。

Answer 1

OnMessage 是一个同步事件，它在收到消息后立即触发，它没有异步替代回调。

如果您想异步处理消息，您可以将生产者与其消费者分离，这可以使用BlockingCollection 来实现，您可以在其中让 SSE 客户端在消息发送后立即捕获消息，例如：

using var bc = new BlockingCollection<ServerEventMessage>();
using var sse = new ServerEventsClient(apiUrl) {
  OnMessage = bc.Add
};
await sse.Connect();
//...

然后，您可以让多个同步和异步消费者按照自己的节奏处理消息，而与使用 Take、TryTake 或 GetConsumingEnumerable API 接收消息的速度无关，例如：

await Task.Run(async () => {
    foreach (var msg in bc.GetConsumingEnumerable())
    {
        // handle msg
        await Task.Delay(100);
    }
});

当您想要停止处理消息时（例如，在处理 ServerEventsClient 时），您可以通过以下方式通知和短路消费者：

bc.CompleteAdding();