Почему IServerStreamWriter не отправляет ответы на уведомления? - PullRequest
Новая
       31

Почему IServerStreamWriter не отправляет ответы на уведомления?

0 голосов
/ 03 мая 2019

Я работаю над кроссплатформенным приложением.Для связи между ними я использую технологию 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;
            }
        }
    }

1 Ответ

0 голосов
/ 03 мая 2019

Ваш notifyObservers метод является асинхронным, но имеет тип возврата void, что означает, что вы не можете его ждать. Вы фактически запускаете метод и возвращаетесь, как только вы нажали первый оператор await, который использует неполное ожидание (первый вызов WriteAsync, вероятно).

Затем вы возвращаете задачу с ReservationResponse, и операция завершается.

Когда этот первый ожидаемый вызов завершится, notifyObservers продолжится, но на этом этапе операция уже завершена, поэтому при попытке записи в поток ответов система выдаст ошибку, которую вы видите.

Я сильно подозреваю, что вы должны вернуть Task из notifyObservers и ждать этого от основного метода ввода: 

// 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);
    }
}
...