Observer pattern using gRPC - C#

Question

Sorry, if this is a stupid question but I don't find any useful information in the internet.

Has anyone ever tried to implement the observer pattern in C# using gRPC as communication? If yes, please show me the link.

Many thanks in advance and best regards.

Answer 1

I have implemented a client convenience class wrapper to turn server streaming calls into regular events for a project I am working. Not sure if this is what you are after. Here is a simple gRPC server that just publishes the time as a string once every second.

syntax = "proto3";
package SimpleTime;

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

message EmptyRequest{}

message TimeResponse
{
   string time = 1;
}

The server implementation, which just loops once a second returning the string representation of the current time until canceled, is as follows

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");
   }
}

For the client, I created a class that contains the gRPC client and exposes the results of the server streaming MonitorTime call as a plain ole .net event.

   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}");
         }
      }
   }

Now create the client and subscribe to the event.

  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}");
  }

Which when run produces

I have left out a lot of error checking and such to make the example smaller. One thing I have done is for gRPC interfaces with many server streaming calls that may or may not be of interest to call clients, is to implement the event accessor (add,remove) to only call the server side streaming method if there is a client that has subscribed to the wrapped event. Hope this is helpful