c# - 创建异步方法并设置其他方法的任务结果

Question

我正在使用使用TCP和SSL流的非常特定的Web API。与API的连接是持久性的(每次写入/读取后都不应关闭)，因为该连接用于接收来自服务器的事件通知。

因此，我实现了循环阅读方法:

private void Listen()
{
    int length;
    do
    {
        var buffer = new byte[TcpClient.ReceiveBufferSize];
        length = SslStream.Read(buffer, 0, TcpClient.ReceiveBufferSize);

        if (length > 0)
        {
            // Process received data
            Listen();
        }
    } while (length > 0);

    //reconnect
}

现在，我需要调用一些API方法。我希望他们支持TPL(异步并等待)。问题在于这些方法的异步部分实际上是在上面的(Listen)的读取方法中实现的。例如:

public async Task<bool> Authenticate(string clientId, string clientSecret)
{
    await SendAuthMessageOverNetwork(clientId, clientSecret);
    return await ReceiveAuthResponseFromServer();
    // The problem is that there is no ReceiveAuthResponseFromServer method - 
    // the response is received in reading thread (Listen).
}

我对TPL不太熟悉。我通过使用一个Task来解决了这个问题，该Task实际上不执行任何操作，而是等待来自读取线程的信号(AutoResetEvent):

private AutoResetEvent _loginEvent;
private bool _loginResult;

public async Task<bool> Authenticate(string clientId, string clientSecret)
{
    await SendAuthMessageOverNetwork(clientId, clientSecret);
    return await Task<bool>.Factory.StartNew(() => { _loginEvent.WaitOne(); return _loginResult; });
}

private void Listen()
{
    ...
    if (msg.Type == MessageTypes.AuthenticationResponse)
    {
        _loginResult = true;
        _loginEvent.Set();
    }
    ...
}

但是我不太喜欢这种解决方案。也许有一种更简单的方法来实现我想要的？可以仅使用Task功能而不使用AutoResetEvent和中间全局变量来完成此操作吗？

Answer 1

没错，使用AutoResetEvent是错误的方法。您实际要做的是:

假设每个SendAuthMessageOverNetwork都由ReceiveAuthResponseFromServer进行匹配，请将它们组合为一个方法。

在该方法中，发送请求并将新的TaskCompletionSource放入队列，该队列可由读取循环

看到。

返回任务完成源的Task属性作为结果

在读取循环中，当您阅读消息时，请从队列中取出下一个完成源，并使用taskSource.SetResult(responseFromServer)

所以像这样:

private readonly Queue<ResponseMessageType> _responseQueue = new Queue<ResponseMessageType>();

public async Task<bool> Authenticate(string clientId, string clientSecret) {
    var response = AsyncRequestAResponse(MakeAuthMessage(clientId, clientSecret));
    return (await response).Type == MessageTypes.AuthenticationResponse
}

public Task<bool> AsyncRequestAResponse(RequestMessageType request) {
    var responseSource = new TaskCompletionSource<ResponseMessageType>();
    _responseQueue.Enqueue(responseSource);
    Send(request);
    return responseSource.Task
}

private void Listen() {
    ...
    if (_responseQueue.Count == 0)
        throw new Exception("Erm, why are they responding before we requested anything?");
    _responseQueue.Dequeue().SetResult(msg);
}

换句话说，使用TaskCompletionSource<T>将内部执行的网络读/写操作转换为公开给调用方的异步操作。

它可能看起来与上面的样子不完全相同...我假设示例中存在继承顺序和一对一的响应/请求。您可能必须将请求ID与响应ID或类似的名称进行匹配，并且超时将把异常而不是结果放入排队的任务中，等等。

另外，如果响应可以与发送的请求同时到达，则在接触队列的操作周围放置锁很重要。