add subscription response

Author: rysweet

dotnet/src/Microsoft.AutoGen/Agents/Services/Grpc/GrpcAgentWorker.cs

@@ -84,6 +84,13 @@ private async Task RunReadPump()
                                 throw new InvalidOperationException($"Failed to register agent: '{message.RegisterAgentTypeResponse.Error}'.");
                             }
                             break;
+                        
+                        case Message.MessageOneofCase.AddSubscriptionResponse:
+                            if (!message.AddSubscriptionResponse.Success)
+                            {
+                                throw new InvalidOperationException($"Failed to add subscription: '{message.AddSubscriptionResponse.Error}'.");
+                            }
+                            break;
 
                         case Message.MessageOneofCase.CloudEvent:
 

dotnet/src/Microsoft.AutoGen/Agents/Services/Grpc/GrpcGateway.cs

@@ -20,6 +20,8 @@ public sealed class GrpcGateway : BackgroundService, IGateway
     // The agents supported by each worker process.
     private readonly ConcurrentDictionary<string, List<GrpcWorkerConnection>> _supportedAgentTypes = [];
     public readonly ConcurrentDictionary<IConnection, IConnection> _workers = new();
+    private readonly ConcurrentDictionary<string, Subscription> _subscriptionsByAgentType = new();
+    private readonly ConcurrentDictionary<string, List<string>> _subscriptionsByTopic = new();
 
     // The mapping from agent id to worker process.
     private readonly ConcurrentDictionary<(string Type, string Key), GrpcWorkerConnection> _agentDirectory = new();
@@ -102,10 +104,22 @@ internal async Task OnReceivedMessageAsync(GrpcWorkerConnection connection, Mess
             case Message.MessageOneofCase.RegisterAgentTypeRequest:
                 await RegisterAgentTypeAsync(connection, message.RegisterAgentTypeRequest);
                 break;
+            case Message.MessageOneofCase.AddSubscriptionRequest:
+                await AddSubscriptionAsync(connection, message.AddSubscriptionRequest);
+                break;
             default:
                 throw new InvalidOperationException($"Unknown message type for message '{message}'.");
         };
     }
+    private async ValueTask AddSubscriptionAsync(GrpcWorkerConnection connection, AddSubscriptionRequest request)
+    {
+        var topic = request.Subscription.TypeSubscription.TopicType;
+        var agentType = request.Subscription.TypeSubscription.AgentType;
+        _subscriptionsByAgentType[agentType] = request.Subscription;
+        _subscriptionsByTopic[topic].Add(agentType);
+        var response = new AddSubscriptionResponse { RequestId = request.RequestId, Error = "", Success = true };
+        await connection.ResponseStream.WriteAsync(response).ConfigureAwait(false);
+    }
     private async ValueTask RegisterAgentTypeAsync(GrpcWorkerConnection connection, RegisterAgentTypeRequest msg)
     {
         connection.AddSupportedType(msg.Type);