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

Question

i在跨平台应用程序上工作。对于它们之间的连接，我使用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返回类型，这意味着您无法等待它。您实际上是 starting 该方法，然后击中第一个使用不完整等待的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);
    }
}