send metadata with channel request

randilfernando · randilfernando · commit c422cf519579 · 2026-01-29T06:40:27.000+05:30
diff --git a/PROTOCOL.md b/PROTOCOL.md
@@ -49,6 +49,9 @@ To open a channel to a service, the client initiates a new HTTP/2 stream with th
 *   `:path`: `/channel`
 *   `x-service-name`: `{local_service_name}` (Name of the service initiating the connection)
 *   `x-session-id`: `{optional_session_id}`
+*   `x-service-name`: `{local_service_name}`
+
+Any additional headers sent during stream initiation or unary calls are exposed to the receiving handler as metadata.
 
 The server accepts the stream by sending headers:
 *   `:status`: `200`
diff --git a/node/README.md b/node/README.md
@@ -43,8 +43,8 @@ Register a service name to handle incoming connections.
 ```typescript
 mesh.registerService('my-service-name')
     // 1. Streaming Handler
-    .onRequestChannel(async (channel) => {
-        console.log(`Accepted connection`);
+    .onRequestChannel(async (channel, metadata) => {
+        console.log(`Accepted connection from session: ${metadata?.['x-session-id']}`);
         try {
             await channel.onData(async (msg) => {
                 console.log('Received:', msg.payload);
@@ -55,8 +55,9 @@ mesh.registerService('my-service-name')
         }
     })
     // 2. Unary Handler (Request/Reply)
-    .onRequestReply(async (msg) => {
+    .onRequestReply(async (msg, metadata) => {
         console.log('Received Unary:', msg.payload);
+        console.log('Metadata:', metadata);
         return { function_name: 'reply', payload: 'Got Unary!' };
     });
 ```
diff --git a/node/src/abstract.ts b/node/src/abstract.ts
@@ -24,12 +24,12 @@ export interface ICommunication {
     // Python ICommunication has: create_channel, listen_channel, start. 
     // It doesn't explicitly have close() in standard steps but Mesh.close might need it.
 
-    createChannel(node: Node): Promise<IChannel>;
-    listenChannel(serviceName: string, callback: (channel: IChannel) => void): void;
+    createChannel(node: Node, metadata?: Record<string, string>): Promise<IChannel>;
+    listenChannel(serviceName: string, callback: (channel: IChannel, metadata?: Record<string, string>) => void): void;
 
     // Unary support
-    sendUnary(node: Node, message: any): Promise<any>;
-    listenUnary(serviceName: string, callback: (msg: any) => Promise<any>): void;
+    sendUnary(node: Node, message: any, metadata?: Record<string, string>): Promise<any>;
+    listenUnary(serviceName: string, callback: (msg: any, metadata?: Record<string, string>) => Promise<any>): void;
 }
 
 export interface IServiceDiscovery {
@@ -63,13 +63,13 @@ export interface IPubSub {
 }
 
 export interface IServiceHandlers {
-    onRequestChannel?: (channel: IChannel) => void;
-    onRequestReply?: (msg: any) => Promise<any>;
+    onRequestChannel?: (channel: IChannel, metadata?: Record<string, string>) => void;
+    onRequestReply?: (msg: any, metadata?: Record<string, string>) => Promise<any>;
 }
 
 export interface IServiceRegistration {
-    onRequestChannel(handler: (channel: IChannel) => void): IServiceRegistration;
-    onRequestReply(handler: (msg: any) => Promise<any>): IServiceRegistration;
+    onRequestChannel(handler: (channel: IChannel, metadata?: Record<string, string>) => void): IServiceRegistration;
+    onRequestReply(handler: (msg: any, metadata?: Record<string, string>) => Promise<any>): IServiceRegistration;
 }
 
 export interface IMesh {
diff --git a/node/src/communication/grpc-communication.ts b/node/src/communication/grpc-communication.ts
@@ -123,8 +123,8 @@ export class GrpcChannel implements IChannel {
 export class GrpcCommunication extends EventEmitter implements ICommunication, IMeshServiceServer {
     private config: MeshConfig;
     private server: grpc.Server;
-    private channelListeners: { [service: string]: (channel: IChannel) => void } = {};
-    private unaryListeners: { [service: string]: (msg: any) => Promise<any> } = {};
+    private channelListeners: { [service: string]: (channel: IChannel, metadata?: Record<string, string>) => void } = {};
+    private unaryListeners: { [service: string]: (msg: any, metadata?: Record<string, string>) => Promise<any> } = {};
 
     // Note: IMeshServiceServer methods are implemented as arrow functions or bound methods usually.
     // Or we simply pass implementation object.
@@ -164,9 +164,18 @@ export class GrpcCommunication extends EventEmitter implements ICommunication, I
         const metadata = call.metadata;
         const serviceName = metadata.get('x-service-name')[0] as string;
 
+        // Convert grpc Metadata to Record<string, string>
+        const metadataRecord: Record<string, string> = {};
+        for (const key in metadata.getMap()) {
+            const val = metadata.get(key);
+            if (val.length > 0) {
+                metadataRecord[key] = val[0] as string;
+            }
+        }
+
         if (serviceName && this.channelListeners[serviceName]) {
             const channel = new GrpcChannel(call);
-            this.channelListeners[serviceName](channel);
+            this.channelListeners[serviceName](channel, metadataRecord);
         } else {
             console.warn(`[gRPC] No channel listener for service: ${serviceName}`);
             call.end();
@@ -177,14 +186,23 @@ export class GrpcCommunication extends EventEmitter implements ICommunication, I
         const metadata = call.metadata;
         const serviceName = metadata.get('x-service-name')[0] as string;
 
+        // Convert grpc Metadata to Record<string, string>
+        const metadataRecord: Record<string, string> = {};
+        for (const key in metadata.getMap()) {
+            const val = metadata.get(key);
+            if (val.length > 0) {
+                metadataRecord[key] = val[0] as string;
+            }
+        }
+
         if (serviceName && this.unaryListeners[serviceName]) {
             const requestMsg = call.request;
             try {
                 const payloadBytes = requestMsg.getPayload_asU8();
                 const jsonString = Buffer.from(payloadBytes).toString('utf8');
                 const msg = JSON.parse(jsonString);
 
-                this.unaryListeners[serviceName](msg)
+                this.unaryListeners[serviceName](msg, metadataRecord)
                     .then(response => {
                         const payload = Buffer.from(JSON.stringify(response), 'utf8');
                         const respMsg = new MeshMessage();
@@ -215,38 +233,50 @@ export class GrpcCommunication extends EventEmitter implements ICommunication, I
         }
     }
 
-    public listenChannel(serviceName: string, callback: (channel: IChannel) => void) {
+    public listenChannel(serviceName: string, callback: (channel: IChannel, metadata?: Record<string, string>) => void) {
         this.channelListeners[serviceName] = callback;
     }
 
-    public listenUnary(serviceName: string, callback: (msg: any) => Promise<any>) {
+    public listenUnary(serviceName: string, callback: (msg: any, metadata?: Record<string, string>) => Promise<any>) {
         this.unaryListeners[serviceName] = callback;
     }
 
-    public async createChannel(node: Node): Promise<IChannel> {
+    public async createChannel(node: Node, metadata?: Record<string, string>): Promise<IChannel> {
         const address = `${node.host}:${node.port}`;
         const client = new MeshServiceClient(address, grpc.credentials.createInsecure());
 
-        const metadata = new grpc.Metadata();
-        metadata.add('x-service-name', node.service_name);
+        const grpcMetadata = new grpc.Metadata();
+        grpcMetadata.add('x-service-name', node.service_name);
 
-        const call = client.send(metadata);
+        if (metadata) {
+            for (const [key, value] of Object.entries(metadata)) {
+                grpcMetadata.add(key, value);
+            }
+        }
+
+        const call = client.send(grpcMetadata);
         return new GrpcChannel(call);
     }
 
-    public async sendUnary(node: Node, message: any): Promise<any> {
+    public async sendUnary(node: Node, message: any, metadata?: Record<string, string>): Promise<any> {
         return new Promise((resolve, reject) => {
             const address = `${node.host}:${node.port}`;
             const client = new MeshServiceClient(address, grpc.credentials.createInsecure());
 
-            const metadata = new grpc.Metadata();
-            metadata.add('x-service-name', node.service_name);
+            const grpcMetadata = new grpc.Metadata();
+            grpcMetadata.add('x-service-name', node.service_name);
+
+            if (metadata) {
+                for (const [key, value] of Object.entries(metadata)) {
+                    grpcMetadata.add(key, value);
+                }
+            }
 
             const payload = Buffer.from(JSON.stringify(message), 'utf8');
             const reqMsg = new MeshMessage();
             reqMsg.setPayload(payload);
 
-            client.unary(reqMsg, metadata, (err: grpc.ServiceError | null, response: MeshMessage) => {
+            client.unary(reqMsg, grpcMetadata, (err: grpc.ServiceError | null, response: MeshMessage) => {
                 if (err) {
                     reject(err);
                 } else {
diff --git a/node/src/mesh.ts b/node/src/mesh.ts
@@ -53,11 +53,11 @@ export class Mesh implements IMesh {
 
     public registerService(serviceName: string): IServiceRegistration {
         const registration = {
-            onRequestChannel: (handler: (channel: IChannel) => void) => {
+            onRequestChannel: (handler: (channel: IChannel, metadata?: Record<string, string>) => void) => {
                 this.communication.listenChannel(serviceName, handler);
                 return registration;
             },
-            onRequestReply: (handler: (msg: any) => Promise<any>) => {
+            onRequestReply: (handler: (msg: any, metadata?: Record<string, string>) => Promise<any>) => {
                 this.communication.listenUnary(serviceName, handler);
                 return registration;
             }
diff --git a/node/src/service-client.ts b/node/src/service-client.ts
@@ -23,7 +23,7 @@ export class ServiceClient implements IServiceClient {
             throw new Error("Failed to map session to node");
         }
 
-        return this.communication.sendUnary(targetNode, message);
+        return this.communication.sendUnary(targetNode, message, { 'x-session-id': sessionId });
     }
 
     public async requestChannel(sessionId: string): Promise<IChannel> {
@@ -40,6 +40,6 @@ export class ServiceClient implements IServiceClient {
         }
 
         console.log(`[Mesh] Mapping session '${sessionId}' -> Node ${targetNode.id}`);
-        return this.communication.createChannel(targetNode);
+        return this.communication.createChannel(targetNode, { 'x-session-id': sessionId });
     }
 }
diff --git a/python/README.md b/python/README.md
@@ -51,8 +51,8 @@ if __name__ == "__main__":
 Register a service name to handle incoming connections.
 
 ```python
-async def stream_handler(channel):
-    print("Accepted connection")
+async def stream_handler(channel, metadata):
+    print(f"Accepted connection from session: {metadata.get('x-session-id')}")
     try:
         async def handler(msg):
             print(f"Received: {msg['payload']}")
@@ -63,8 +63,9 @@ async def stream_handler(channel):
     except Exception:
         pass
 
-async def unary_handler(msg):
+async def unary_handler(msg, metadata):
     print(f"Received Unary: {msg['payload']}")
+    print(f"Metadata: {metadata}")
     return {"function_name": "reply", "payload": "Got Unary!"}
 
 mesh.register_service("my-service-name") \
diff --git a/python/src/mesh/abstract.py b/python/src/mesh/abstract.py
@@ -1,10 +1,10 @@
 from abc import ABC, abstractmethod
-from typing import Callable, Awaitable, List, Optional, Any, TypedDict, Union
+from typing import Callable, Awaitable, List, Optional, Any, TypedDict, Union, Dict
 from mesh.shared.types import Node
 
 class IServiceHandlers(TypedDict, total=False):
-    on_request_channel: Optional[Callable[['IChannel'], Awaitable[None]]]
-    on_request_reply: Optional[Callable[[Any], Awaitable[Any]]]
+    on_request_channel: Optional[Callable[['IChannel', Dict[str, str]], Awaitable[None]]]
+    on_request_reply: Optional[Callable[[Any, Dict[str, str]], Awaitable[Any]]]
 
 class IServiceClient(ABC):
     @abstractmethod
@@ -104,27 +104,38 @@ async def on_data(self, handler: Callable[[Any], Awaitable[None]]):
 
 class ICommunication(ABC):
     @abstractmethod
-    async def create_channel(self, target: Union[str, Node]) -> IChannel:
+    async def create_channel(self, target: Union[str, Node], metadata: Dict[str, str] = None) -> IChannel:
         """
         Open a new channel to a service or specific node.
         """
         pass
 
     @abstractmethod
-    def listen_channel(self, service_name: str, callback: Callable[[IChannel], Awaitable[None]]):
+    def listen_channel(self, service_name: str, callback: Callable[[IChannel, Dict[str, str]], Awaitable[None]]):
         """
         Register a callback to handle incoming channels for a service.
         The callback receives a IChannel object to read/write messages.
         """
         pass
 
     @abstractmethod
-    async def start(self):
+    async def send_unary(self, node: Node, message: Any, metadata: Dict[str, str] = None) -> Any:
+        """
+        Send a unary message to a node.
+        """
+        pass
+
+    @abstractmethod
+    def listen_unary(self, service_name: str, callback: Callable[[Any, Dict[str, str]], Awaitable[Any]]):
         """
-        Start the communication server.
+        Register a callback to handle unary messages.
         """
         pass
 
+    @abstractmethod
+    async def start(self):
+        pass
+
 class IServiceDiscovery(ABC):
     @abstractmethod
     def register_service(self, node: Node):
@@ -151,11 +162,11 @@ def get_service(self, service_name: str) -> List[Node]:
 
 class IServiceRegistration(ABC):
     @abstractmethod
-    def on_request_channel(self, handler: Callable[['IChannel'], Awaitable[None]]) -> 'IServiceRegistration':
+    def on_request_channel(self, handler: Callable[['IChannel', Dict[str, str]], Awaitable[None]]) -> 'IServiceRegistration':
         pass
     
     @abstractmethod
-    def on_request_reply(self, handler: Callable[[Any], Awaitable[Any]]) -> 'IServiceRegistration':
+    def on_request_reply(self, handler: Callable[[Any, Dict[str, str]], Awaitable[Any]]) -> 'IServiceRegistration':
         pass
 
 class IMesh(ABC):
diff --git a/python/src/mesh/communication/grpc_communication.py b/python/src/mesh/communication/grpc_communication.py
@@ -98,7 +98,7 @@ async def Send(self, request_iterator, context):
             channel = GrpcChannel(request_iterator, outgoing_queue)
             
             # Notify listener
-            asyncio.create_task(self.comm.listeners[service_name](channel))
+            asyncio.create_task(self.comm.listeners[service_name](channel, metadata))
 
             # Stream outgoing messages back to client
             while True:
@@ -118,7 +118,7 @@ async def Unary(self, request, context):
         if service_name and service_name in self.comm.unary_listeners:
              try:
                  obj = json.loads(request.payload.decode('utf-8'))
-                 response_obj = await self.comm.unary_listeners[service_name](obj)
+                 response_obj = await self.comm.unary_listeners[service_name](obj, metadata)
                  
                  payload = json.dumps(response_obj).encode('utf-8')
                  return mesh_pb2.MeshMessage(payload=payload)
@@ -137,8 +137,9 @@ class GrpcCommunication(ICommunication):
     def __init__(self, config: MeshConfig, service_discovery: IServiceDiscovery):
         self.config = config
         self.service_discovery = service_discovery
-        self.listeners: Dict[str, Callable[[IChannel], Awaitable[None]]] = {}
-        self.unary_listeners: Dict[str, Callable[[Any], Awaitable[Any]]] = {}
+        self.service_discovery = service_discovery
+        self.listeners: Dict[str, Callable[[IChannel, Dict[str, str]], Awaitable[None]]] = {}
+        self.unary_listeners: Dict[str, Callable[[Any, Dict[str, str]], Awaitable[Any]]] = {}
         self._server = None
 
     async def start(self):
@@ -163,13 +164,13 @@ async def start(self):
         # But if main program exits, it stops. 
         # Mesh start() usually returns. The server keeps running attached to loop.
 
-    def listen_channel(self, service_name: str, callback: Callable[[IChannel], Awaitable[None]]):
+    def listen_channel(self, service_name: str, callback: Callable[[IChannel, Dict[str, str]], Awaitable[None]]):
         self.listeners[service_name] = callback
 
-    def listen_unary(self, service_name: str, callback: Callable[[Any], Awaitable[Any]]):
+    def listen_unary(self, service_name: str, callback: Callable[[Any, Dict[str, str]], Awaitable[Any]]):
         self.unary_listeners[service_name] = callback
 
-    async def create_channel(self, target: Union[str, Node]) -> IChannel:
+    async def create_channel(self, target: Union[str, Node], metadata: Dict[str, str] = None) -> IChannel:
         if isinstance(target, str):
             service_name = target
             nodes = self.service_discovery.get_service(service_name)
@@ -193,22 +194,28 @@ async def request_generator():
                      return
                 yield msg
         
-        metadata = [('x-service-name', node.service_name)]
+        metadata_list = [('x-service-name', node.service_name)]
+        if metadata:
+            for k, v in metadata.items():
+                metadata_list.append((k, v))
         
-        response_iterator = stub.Send(request_generator(), metadata=metadata)
+        response_iterator = stub.Send(request_generator(), metadata=metadata_list)
         
         return GrpcChannel(response_iterator, outgoing_queue)
 
-    async def send_unary(self, node: Node, message: Any) -> Any:
+    async def send_unary(self, node: Node, message: Any, metadata: Dict[str, str] = None) -> Any:
         channel_creds = grpc.aio.insecure_channel(f'{node.host}:{node.port}')
         stub = mesh_pb2_grpc.MeshServiceStub(channel_creds)
         
         try:
              payload = json.dumps(message).encode('utf-8')
              req = mesh_pb2.MeshMessage(payload=payload)
-             metadata = [('x-service-name', node.service_name)]
+             metadata_list = [('x-service-name', node.service_name)]
+             if metadata:
+                for k, v in metadata.items():
+                    metadata_list.append((k, v))
              
-             resp = await stub.Unary(req, metadata=metadata)
+             resp = await stub.Unary(req, metadata=metadata_list)
              
              obj = json.loads(resp.payload.decode('utf-8'))
              return obj
diff --git a/python/src/mesh/mesh.py b/python/src/mesh/mesh.py
diff --git a/python/src/mesh/service_client.py b/python/src/mesh/service_client.py

Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ export class ServiceClient implements IServiceClient {`
`23`	`23`	`throw new Error("Failed to map session to node");`
`24`	`24`	`}`
`25`	`25`
`26`		`- return this.communication.sendUnary(targetNode, message);`
	`26`	`+ return this.communication.sendUnary(targetNode, message, { 'x-session-id': sessionId });`
`27`	`27`	`}`
`28`	`28`
`29`	`29`	`public async requestChannel(sessionId: string): Promise<IChannel> {`
`@@ -40,6 +40,6 @@ export class ServiceClient implements IServiceClient {`
`40`	`40`	`}`
`41`	`41`
`42`	`42`	console.log(`[Mesh] Mapping session '${sessionId}' -> Node ${targetNode.id}`);
`43`		`- return this.communication.createChannel(targetNode);`
	`43`	`+ return this.communication.createChannel(targetNode, { 'x-session-id': sessionId });`
`44`	`44`	`}`
`45`	`45`	`}`