title: gRPC Push and Fan-Out
body: <p>Is it possible and if so good practice to use gRPC as a push service with fan-out functionality?
  In the examples given from Google, there is following code for the server side (C#):</p>
  
  <pre><code>    public override async Task ListFeatures(Rectangle request, IServerStreamWriter&lt;Feature&gt; responseStream, ServerCallContext context)
      {
          var responses = features.FindAll( (feature) =&gt; feature.Exists() &amp;&amp; request.Contains(feature.Location) );
          foreach (var response in responses)
          {
              await responseStream.WriteAsync(response);
          }
      }
  </code></pre>
  
  <p>The problem is here that:</p>
  
  <ol>
  <li>Data is only generated and written if the client explicitly asks for it.</li>
  <li>Only the client that asked will get the new data.</li>
  </ol>
  
  <p>I think what I need is:</p>
  
  <ol>
  <li>Keep all the IServerStreamWriters for each client asked (subscribed).</li>
  <li>Trigger the write by an external event when new data is available.</li>
  <li>Write to all streamWriters</li>
  </ol>
  
  <p><strong>Edit:</strong>
  Based on Carl's suggestions I now have the following:</p>
  
  <p>Proto:</p>
  
  <pre><code>service PubSub {
   rpc Subscribe(Subscription) returns (stream Event) {}
   rpc Unsubscribe(Subscription) returns (Unsubscription) {}
  }
  
  message Event
  {
     string Value = 1;
  }
  message Subscription
  {
    string Id = 1;
  }
  message Unsubscription
  {
    string Id = 1;
  }
  </code></pre>
  
  <p>PubSubImpl:</p>
  
  <pre><code>public class PubSubImpl : PubSub.PubSubBase
  {
      private readonly BufferBlock&lt;Event&gt; _buffer = new BufferBlock&lt;Event&gt;();
  
      private Dictionary&lt;string, IServerStreamWriter&lt;Event&gt;&gt; _subscriberWritersMap =
          new Dictionary&lt;string, IServerStreamWriter&lt;Event&gt;&gt;();
  
       public override async Task Subscribe(Subscription subscription, IServerStreamWriter&lt;Event&gt; responseStream, ServerCallContext context)
      {
          //Dict to hold a streamWriter for each subscriber.
          _subscriberWritersMap[subscription.Id] = responseStream;
  
          while (_subscriberWritersMap.ContainsKey(subscription.Id))
          {
              //Wait on BufferBlock from MS Dataflow package.
              var @event = await _buffer.ReceiveAsync();
              foreach (var serverStreamWriter in _subscriberWritersMap.Values)
              {
                  await serverStreamWriter.WriteAsync(@event);
              }
          }
      }
  
      public override Task&lt;Unsubscription&gt; Unsubscribe(Subscription request, ServerCallContext context)
      {
          _subscriberWritersMap.Remove(request.Id);
          return Task.FromResult(new Unsubscription() { Id = request.Id });
      }
  
      public void Publish(string input)
      {
          _buffer.Post(new Event() { Value = input });
      }
  }
  </code></pre>
  
  <p>"Push"s can now be sent like this:</p>
  
  <pre><code>   while ((input = Console.ReadLine()) != "q")
      {
        pubsubImp.Publish(input);
      }
  </code></pre>
  
  <p>On client side I have:</p>
  
  <pre><code>public async Task Subscribe()
  {
      _subscription = new Subscription() { Id = Guid.NewGuid().ToString() };
      using (var call = _pubSubClient.Subscribe(_subscription))
      {
          //Receive
          var responseReaderTask = Task.Run(async () =&gt;
          {
              while (await call.ResponseStream.MoveNext())
              {
                  Console.WriteLine("Event received: " + call.ResponseStream.Current);
              }
          });
  
          await responseReaderTask;
      }
  }
  
  public void Unsubscribe()
  {
      _pubSubClient.Unsubscribe(_subscription);
  }
  </code></pre>
  
  <p>Client-Main works like this:</p>
  
  <pre><code>static void Main(string[] args)
  {
      var channel = new Channel("127.0.0.1:50052", 
                                  ChannelCredentials.Insecure);
      var subscriber = new Subsriber(new PubSub.PubSubClient(channel));
  
      Task.Run(async () =&gt;
      {
          await subscriber.Subscribe();
      }).GetAwaiter();
  
      Console.WriteLine("Hit key to unsubscribe");
      Console.ReadLine();
  
      subscriber.Unsubscribe();
  
      Console.WriteLine("Unsubscribed...");
  
      Console.WriteLine("Hit key to exit...");
      Console.ReadLine();
  
  }
  </code></pre>
  
  <p>At the moment it looks like it works. Is that how it should/could be done?
  Test solution can be found at:
  <a href="https://github.com/KingKnecht/gRPC-PubSub" rel="noreferrer"><a href="https://github.com/KingKnecht/gRPC-PubSub" rel="noreferrer">https://github.com/KingKnecht/gRPC-PubSub</a></a></p>