使用gRPC的观察者模式-C＃

Question

抱歉，如果这是一个愚蠢的问题，但是我在互联网上找不到任何有用的信息。

有没有人尝试使用gRPC作为通信在C＃中实现观察者模式？如果是，请给我显示链接。

非常感谢和最诚挚的问候。

Answer 1

我已经实现了一个客户端便利类包装器，以将服务器流调用转换为我正在工作的项目的常规事件。不确定这是否是您要的。这是一个简单的gRPC服务器，它每秒将时间以字符串形式发布一次。

syntax = "proto3";
package SimpleTime;

service SimpleTimeService
{
   rpc MonitorTime(EmptyRequest) returns (stream TimeResponse);
}

message EmptyRequest{}

message TimeResponse
{
   string time = 1;
}

服务器实现如下，每秒循环一次，返回当前时间的字符串表示，直到取消为止。

public override async Task MonitorTime(EmptyRequest request, IServerStreamWriter<TimeResponse> responseStream, ServerCallContext context)
{
   try
   {
      while (!context.CancellationToken.IsCancellationRequested)
      {
         var response = new TimeResponse
         {
            Time = DateTime.Now.ToString()
         };
         await responseStream.WriteAsync(response);
         await Task.Delay(1000);
      }
   }
   catch (Exception)
   { 
      Console.WriteLine("Exception on Server");
   }
}

对于客户端，我创建了一个包含gRPC客户端的类，并将服务器流MonitorTime调用的结果显示为纯ole .net事件。

   public class SimpleTimeEventClient
   {
      private SimpleTime.SimpleTimeService.SimpleTimeServiceClient mClient = null;
      private CancellationTokenSource mCancellationTokenSource = null;
      private Task mMonitorTask = null;
      public event EventHandler<string> OnTimeReceived;

      public SimpleTimeEventClient()
      {
         Channel channel = new Channel("127.0.0.1:50051", ChannelCredentials.Insecure);
         mClient = new SimpleTime.SimpleTimeService.SimpleTimeServiceClient(channel);
      }

      public void Startup()
      {
         mCancellationTokenSource = new CancellationTokenSource();
         mMonitorTask = Task.Run(() => MonitorTimeServer(mCancellationTokenSource.Token));
      }

      public void Shutdown()
      {
         mCancellationTokenSource.Cancel();
         mMonitorTask.Wait(10000);
      }

      private async Task MonitorTimeServer(CancellationToken token)
      {
         try
         {
            using (var call = mClient.MonitorTime(new SimpleTime.EmptyRequest()))
            {
               while(await call.ResponseStream.MoveNext(token))
               {
                  var timeResult = call.ResponseStream.Current;
                  OnTimeReceived?.Invoke(this, timeResult.Time);
               }
            }
         }
         catch(Exception e)
         {
            Console.WriteLine($"Exception encountered in MonitorTimeServer:{e.Message}");
         }
      }
   }

现在创建客户端并预订活动。

  static void Main(string[] args)
  {
     SimpleTimeEventClient client = new SimpleTimeEventClient();
     client.OnTimeReceived += OnTimeReceivedEventHandler;
     client.Startup();
     Console.WriteLine("Press any key to exit");
     Console.ReadKey();
     client.Shutdown();

  }

  private static void OnTimeReceivedEventHandler(object sender, string e)
  {
     Console.WriteLine($"Time: {e}");
  }

运行时会产生哪个

我遗漏了很多错误检查，因此使示例更小。我所做的一件事是对于带有许多服务器流调用的gRPC接口，这些调用可能不涉及调用客户端，而是实现事件访问器（添加，删除），以便仅在有客户端的情况下才调用服务器端流方法订阅了包装事件。希望对您有所帮助