0

我正在努力使用 NuGet 包 websocket-client ( https://github.com/Marfusios/websocket-client )的观察者模式

与 WebSocket 服务器的连接稳定且正在运行。

每个请求在负载中都有一个请求 ID。客户端将其发送到服务器，服务器以 ID 和真实数据进行响应。

在客户端，我需要将每个响应分配给相应的请求。

我以为我可以这样做：

public Task<Data> GetDataAsync()
{
    var webSocket = new WebsocketClient(Uri);
    await webSocket.Start();

    var requestId = Guid.NewGuid();

    var tcs = new TaskCompletionSource<Data>();

    var disposable = webSocket
    .MessageReceived
    .Where(message => message.Text.Contains(requestId))
    .Subscribe(message=>
    {
        var data = ParseData(message.Text);
        tcs.SetResult(data);
    });

    return tcs.Task;
}

但它实际上从不跳转到 subscribe 方法。我用错了吗？

4

2 回答 2

0

我认为

public Task<Data> GetDataAsync(string request)
{
    var requestId = Guid.NewGuid().ToString();

    var responseTask = WebSocket
    .MessageReceived
    .Timeout(TimeSpan.FromSeconds(5))
    .FirstOrDefaultAsync(message => message.Text.Contains(requestId));

    WebSocket.Send(request);

    var responseMessage = await responseTask;

    return ParseMessage(responseMessage);
}

是要走的路。我什至更喜欢SingleOrDefaultAsync而不是FirstOrDefaultAsync因为只有一条带有该请求 ID 的消息。但这不起作用。它总是在超时时运行。

于 2021-02-16T15:59:26.570 回答
-1

您使您的代码比我需要的要复杂得多。Rx 让我们等待一个 observable 来获取最后产生的值。您可以这样编写代码：

public async Task<Data> GetDataAsync() =>
    await
        Observable
            .Using(
                () => new WebsocketClient(Uri),
                ws =>
                    from x in Observable.FromAsync(() => ws.Start())
                    let requestId = Guid.NewGuid()
                    from m in ws.MessageReceived
                    where m.Text.Contains(requestId)
                    select ParseData(m.Text))
            .Take(1)
            .Timeout(TimeSpan.FromSeconds(5.0));
于 2021-02-15T00:11:46.000 回答