为什么 IServerStreamWriter 不发送通知响应？

Question

我在一个跨平台的应用程序上工作。对于它们之间的连接，我使用 gRPC 技术。当客户端连接到服务器时，它会被添加到位于服务器实现中的观察者列表中。当一个客户端连接时，我想向其他连接的客户端发送一条消息，告诉他们有一个新的客户端连接。问题是，当我想使用列表中的观察者向新客户端连接的客户端发送响应时，出现以下异常：

Grpc.Core.RpcException: 'Status(StatusCode=Unknown, Detail="Exception was thrown by handler.")'

这是我声明服务器的原始文件：

syntax = "proto3";

package com.example.grpc.chat;

message ChatMessage {
    string from = 1;
    string message = 2;
}

message ChatMessageFromServer {
    ChatMessage message = 2;
}

service ChatService {
    rpc Login(ChatMessage ) returns (stream ChatMessageFromServer);
}

服务器代码：

 public class ChatServiceImpl : ChatService.ChatServiceBase
    {
        private static HashSet<IServerStreamWriter<ChatMessageFromServer>> responseStreams = new HashSet<IServerStreamWriter<ChatMessageFromServer>>();

        /*
         * if the stream object (from "for" statement inside this method) isn't the responseStream object given in the list with parameters,
         * the rest of clients aren't notified when a new login request is pushed.
         */
        public override async Task Login(global::Com.Example.Grpc.Chat.ChatMessage request,
            IServerStreamWriter<global::Com.Example.Grpc.Chat.ChatMessageFromServer> responseStream, 
            ServerCallContext context)
        {
            Console.WriteLine("Login method from server");
            responseStreams.Add(responseStream);

            // Create a server message that wraps the client message
            var message = new ChatMessageFromServer
            {
                Message = new ChatMessage
                {
                    From = "login",
                    Message = "hello"
                }
            };
            // If stream variable isn't equal to responseStream from list of parameters, the client corresponding to that stream isn't notified and it's thrown the above exception
            foreach (var stream in responseStreams)
            {
                await stream.WriteAsync(message);
            }
        }

    }

客户端发送登录请求的客户端代码：

public partial class ChatForm : Form
    {
        private const string Host = "localhost";
        private const int Port = 9090;

        private ChatService.ChatServiceClient _chatService;

        public ChatForm()
        {
            //InitializeComponent();
            InitializeGrpc();
        }

        private void InitializeGrpc()
        {
            // Create a channel
            var channel = new Channel(Host + ":" + Port, ChannelCredentials.Insecure);

            // Create a client with the channel
            _chatService = new ChatService.ChatServiceClient(channel);
        }

        private async void ChatForm_Load(object sender, EventArgs e)
        {

            var message = new ChatMessage
            {
                From = "Unknown",
                Message = "Login text"
            };

            // Open a connection to the server
            try
            {
                using (var call = _chatService.Login(message))
                {
                    // Read messages from the response stream
                    while (await call.ResponseStream.MoveNext(CancellationToken.None))
                    {
                        var serverMessage = call.ResponseStream.Current;
                        var otherClientMessage = serverMessage.Message;
                        var displayMessage = string.Format("{0}:{1}{2}", otherClientMessage.From, otherClientMessage.Message, Environment.NewLine);
                        chatTextBox.Text += displayMessage;
                    }

                }
            }
            catch (RpcException )
            {

                throw;
            }
        }
    }

Answer 1

您的notifyObservers 方法是异步的，但有一个void 返回类型，这意味着您不能等待它。您实际上是在开始该方法，并在您点击第一个使用不完整等待的await 运算符（可能是第一个WriteAsync 调用）后立即返回。

然后您返回带有ReservationResponse 的任务，操作完成。

当第一个等待调用完成时，notifyObservers 将继续，但此时操作已经完成，因此当您尝试写入响应流时，系统将抛出您看到的错误。

我强烈怀疑你应该从notifyObservers 返回一个Task 并从你的主入口方法中等待它：

// Names changed to be conventional C#
public override async Task<ReservationResponse> SaveReservation(
    global::Res.Protocol.ReservationRequest request, ServerCallContext context)
{
    // some code for saving my reservation in repository database
    ReservationResponse response = new Res.Protocol.ReservationResponse
    {
        Type = ReservationResponse.Types.Type.Savereservation,
        Journey = GetProtoJourney(journey)
    };

    await NotifyObserversAsync(response);

    // Note: no Task.FromResult, as you're in an async method. The response
    // will already be wrapped in a task.
    return new ReservationResponse
    {
        Type = ReservationResponse.Types.Type.Savereservation
    };
}

public async Task NotifyObserversAsync(Res.Protocol.ReservationResponse response)
{
    foreach (var ob in responseStreams)
    {
        await ob.WriteAsync(response);
    }
}