如何在 gRPC 上实现服务器“回调”答案

【问题标题】：How to implement a server "callback" on gRPC如何在 gRPC 上实现服务器“回调”
【发布时间】：2021-06-01 21:21:36
【问题描述】：

我在 gRPC 上创建了一个 POC，看看它可以做什么并在 .Net 5 上迁移 WCF 应用程序

POC 是一个简单的聊天，发送新消息，所有客户端都会收到消息

共享代码

public interface IChatService
{
    IAsyncEnumerable<MessageResponse> Subscribe();
    ValueTask PostMessage(MessageRequest msg);
}

[ProtoContract]
public class MessageRequest
{
    public MessageRequest(string user, string msg)
    {
        User = user;
        Msg = msg;
    }

    [ProtoMember(1)]
    public string Msg { get; }

    [ProtoMember(2)]
    public string User { get; }
}

[ProtoContract]
public class MessageResponse
{
    public MessageResponse(string user, string msg)
    {
        User = user;
        Msg = msg;
        Date = DateTime.Now;
    }

    [ProtoMember(1)]
    public string Msg { get; }

    [ProtoMember(2)]
    public string User { get; }

    [ProtoMember(2)]
    public DateTime Date { get; }
}

客户端代码：

        using var http = GrpcChannel.ForAddress("http://localhost:10042");

        var chatService = http.CreateGrpcService<IChatService>();

        await foreach (var msg in chatService.Subscribe())
        {
            Console.WriteLine($"{msg.Date} : {msg.User} - {msg.Msg}");
        }

        await chatService.PostMessage(new MessageRequest("User1", "MyMessage"));

服务器端代码：

public class ChatService : IChatService
{
    private static readonly List<MessageResponse> OldMessages = new List<MessageResponse>();

    public async IAsyncEnumerable<MessageResponse> Subscribe()
    {
        foreach (var msg in OldMessages)
        {
            await Task.Delay(10);
            yield return msg;
        }
    }

    public async ValueTask PostMessage(MessageRequest msg)
    {
        await Task.Delay(10);
        OldMessages.Add(new MessageResponse(msg.User, msg.Msg));
    }
}

在想法中，订阅方法在发布新消息时发送给客户端和 PostMessage 方法在聊天中添加消息（很明显）

但我不明白如何在服务器端代码中将消息“添加”到 Subscribe 的 IAsyncEnumerable 返回？使用 proto，它会是这样的：

service ChatService {
  rpc Subscribe (request) returns (stream response);
}

感谢您的帮助！

【问题讨论】：

标签： .net grpc protobuf-net

【解决方案1】：

目前，您正在对列表使用简单的foreach 循环。如果您将.Add 加入此列表，foreach 将中断，并且不会返回更多数据。您本质上需要的是为生产者/消费者场景设计的线程安全异步队列机制。幸运的是，Channel<T> 提供了这样的机制。但是，您仍然需要将数据推送到所有订阅者实例中，这可能是更大的挑战 - 通常情况下，您不想通知调用 .Add 的同一个实例

【讨论】：