removed message class, can send any via channel

Author: randilfernando

node/mesh_peer.ts

@@ -47,9 +47,9 @@ async function main() {
                 const sendLoop = async () => {
                     let counter = 0;
                     while (true) {
-                        const payload = `Message from Node #${counter}`;
-                        await channel.send({ function_name: "stream", payload });
-                        console.log(`[Node] Sent: ${payload}`);
+                        const payload = { text: `Message from Node #${counter}`, count: counter };
+                        await channel.send(payload);
+                        console.log(`[Node] Sent:`, payload);
                         counter++;
                         await new Promise(r => setTimeout(r, 1000)); // Send every 1s
                     }

node/src/abstract.ts

@@ -1,10 +1,10 @@
-import { Node, Message } from './shared/types';
+import { Node } from './shared/types';
 // Removed import { Channel } from './communication/h2-communication' to avoid circular dependency
 // Channel is now defined here.
 // User requested interfaces start with I.
 export interface IChannel {
-    send(message: Message): Promise<void>;
-    receive(): Promise<Message>;
+    send(message: any): Promise<void>;
+    receive(): Promise<any>;
     close(): Promise<void>;
     // onClose is specific to H2 implementation? Python abstract Channel doesn't have it.
     // Python abstract Channel has: send, receive, close.
@@ -14,7 +14,7 @@ export interface IChannel {
     // "but python abstraction has more interfaces than node"
     // I will include onClose as it is useful for Node.
     onClose?(handler: () => void): void;
-    [Symbol.asyncIterator](): AsyncIterator<Message>;
+    [Symbol.asyncIterator](): AsyncIterator<any>;
 }
 
 export interface ICommunication {
@@ -27,8 +27,8 @@ export interface ICommunication {
     listenChannel(serviceName: string, callback: (channel: IChannel) => void): void;
 
     // Unary support
-    sendUnary(node: Node, message: Message): Promise<Message>;
-    listenUnary(serviceName: string, callback: (msg: Message) => Promise<Message>): void;
+    sendUnary(node: Node, message: any): Promise<any>;
+    listenUnary(serviceName: string, callback: (msg: any) => Promise<any>): void;
 }
 
 export interface IServiceDiscovery {
@@ -38,7 +38,7 @@ export interface IServiceDiscovery {
 }
 
 export interface IServiceClient {
-    requestReply(sessionId: string, message: Message): Promise<Message>;
+    requestReply(sessionId: string, message: any): Promise<any>;
     requestChannel(sessionId: string): Promise<IChannel>;
 }
 
@@ -63,7 +63,7 @@ export interface IPubSub {
 
 export interface IServiceHandlers {
     onRequestChannel?: (channel: IChannel) => void;
-    onRequestReply?: (msg: Message) => Promise<Message>;
+    onRequestReply?: (msg: any) => Promise<any>;
 }
 
 export interface IMesh {

node/src/communication/h2-communication.ts

@@ -1,6 +1,6 @@
 import * as http2 from 'http2';
 import { EventEmitter } from 'events';
-import { Node, MeshConfig, Message } from '../shared/types';
+import { Node, MeshConfig } from '../shared/types';
 
 import { IChannel, ICommunication } from '../abstract';
 
@@ -9,8 +9,8 @@ import { IChannel, ICommunication } from '../abstract';
 export class H2Channel implements IChannel {
     private stream: http2.ClientHttp2Stream | http2.ServerHttp2Stream;
     // We need a queue for received messages because receive() calls might not align with data events
-    private receiveQueue: Message[] = [];
-    private receiveResolvers: Array<{ resolve: (msg: Message) => void, reject: (err: Error) => void }> = [];
+    private receiveQueue: any[] = [];
+    private receiveResolvers: Array<{ resolve: (msg: any) => void, reject: (err: Error) => void }> = [];
     private buffer: Buffer = Buffer.alloc(0);
     private isClosed: boolean = false;
     private closeHandlers: (() => void)[] = [];
@@ -58,7 +58,7 @@ export class H2Channel implements IChannel {
 
                 try {
                     const msgString = payloadBuf.toString('utf8');
-                    const msg: Message = JSON.parse(msgString);
+                    const msg = JSON.parse(msgString);
                     this._pushMessage(msg);
                 } catch (e) {
                     console.error("Failed to parse message", e);
@@ -69,7 +69,7 @@ export class H2Channel implements IChannel {
         }
     }
 
-    private _pushMessage(msg: Message) {
+    private _pushMessage(msg: any) {
         if (this.receiveResolvers.length > 0) {
             const { resolve } = this.receiveResolvers.shift()!;
             resolve(msg);
@@ -78,7 +78,7 @@ export class H2Channel implements IChannel {
         }
     }
 
-    public async send(message: Message): Promise<void> {
+    public async send(message: any): Promise<void> {
         if (this.isClosed) {
             throw new Error("Channel closed");
         }
@@ -95,7 +95,7 @@ export class H2Channel implements IChannel {
         });
     }
 
-    public async receive(): Promise<Message> {
+    public async receive(): Promise<any> {
         if (this.receiveQueue.length > 0) {
             return this.receiveQueue.shift()!;
         }
@@ -122,7 +122,7 @@ export class H2Channel implements IChannel {
         }
     }
 
-    public async *[Symbol.asyncIterator](): AsyncIterator<Message> {
+    public async *[Symbol.asyncIterator](): AsyncIterator<any> {
         while (!this.isClosed) {
             try {
                 const msg = await this.receive();
@@ -141,7 +141,7 @@ export class H2Communication extends EventEmitter implements ICommunication {
     private config: MeshConfig;
     private server?: http2.Http2Server;
     private channelListeners: { [service: string]: (channel: IChannel) => void } = {};
-    private unaryListeners: { [service: string]: (msg: Message) => Promise<Message> } = {};
+    private unaryListeners: { [service: string]: (msg: any) => Promise<any> } = {};
 
     constructor(config: MeshConfig) {
         super();
@@ -213,7 +213,7 @@ export class H2Communication extends EventEmitter implements ICommunication {
         this.channelListeners[serviceName] = callback;
     }
 
-    public listenUnary(serviceName: string, callback: (msg: Message) => Promise<Message>) {
+    public listenUnary(serviceName: string, callback: (msg: any) => Promise<any>) {
         this.unaryListeners[serviceName] = callback;
     }
 
@@ -233,7 +233,7 @@ export class H2Communication extends EventEmitter implements ICommunication {
         });
     }
 
-    public async sendUnary(node: Node, message: Message): Promise<Message> {
+    public async sendUnary(node: Node, message: any): Promise<any> {
         return new Promise((resolve, reject) => {
             const client = http2.connect(`http://${node.host}:${node.port}`);
             client.on('error', (err) => reject(err));

node/src/mesh.ts

@@ -1,13 +1,13 @@
 import { v4 as uuidv4 } from 'uuid';
 import * as os from 'os';
-import { MeshConfig, Node, Message } from './shared/types';
+import { MeshConfig, Node } from './shared/types';
 import { RedisServiceDiscovery } from './discovery/redis-discovery';
 import { H2Communication } from './communication/h2-communication';
 import { ServiceClient } from './service-client';
 import { IMesh, IServiceClient, ITopic, IServiceHandlers, IChannel, IServiceDiscovery, ICommunication, IPubSub } from './abstract';
 import { RedisPubSub, Topic } from './communication/pubsub';
 
-export { MeshConfig, Node, Message, IChannel, IServiceHandlers, IChannel as Channel, IServiceHandlers as ServiceHandlers, Topic }; // Export both for compat
+export { MeshConfig, Node, IChannel, IServiceHandlers, IChannel as Channel, IServiceHandlers as ServiceHandlers, Topic }; // Export both for compat
 // User asked to standardize naming. I'll export renamed IChannel and IServiceHandlers.
 // IF old code relies on Channel/ServiceHandlers export from Mesh, I should alias them or update consumer code.
 // Given "standardize abstraction", ideally I use IChannel everywhere.

node/src/service-client.ts

@@ -1,4 +1,4 @@
-import { Message } from './shared/types';
+import { Node } from './shared/types';
 import { IChannel, ICommunication, IServiceDiscovery } from './abstract';
 import { ConsistentHash } from './shared/hashing';
 import { IServiceClient } from './abstract';
@@ -10,7 +10,7 @@ export class ServiceClient implements IServiceClient {
         private communication: ICommunication
     ) { }
 
-    public async requestReply(sessionId: string, message: Message): Promise<Message> {
+    public async requestReply(sessionId: string, message: any): Promise<any> {
         const nodes = await this.discovery.getService(this.serviceName);
         if (nodes.length === 0) {
             throw new Error(`No nodes found for service '${this.serviceName}'`);

node/src/shared/types.ts

@@ -5,10 +5,7 @@ export interface Node {
     service_name: string;
 }
 
-export interface Message {
-    function_name: string;
-    payload: any;
-}
+
 
 // Configuration Interfaces
 export interface RedisConfig {

python/mesh_peer.py

@@ -2,7 +2,7 @@
 import logging
 import random
 from mesh_lib import Mesh
-from mesh_lib.shared.types import MeshConfig, ServiceDiscoveryConfig, Message
+from mesh_lib.shared.types import MeshConfig, ServiceDiscoveryConfig
 
 # --- Configuration & Setup ---
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - [Python] - %(message)s')
@@ -37,7 +37,7 @@ async def send_loop():
             try:
                 while True:
                     msg_payload = f"Message from Python #{counter}"
-                    out_msg = Message(function_name="stream", payload=msg_payload)
+                    out_msg = {"text": msg_payload, "count": counter}
                     await channel.send(out_msg)
                     logger.info(f"Sent: {msg_payload}")
                     counter += 1
@@ -49,7 +49,7 @@ async def send_loop():
         async def receive_loop():
             try:
                 async for msg in channel:
-                    logger.info(f"Received: {msg.payload}")
+                    logger.info(f"Received: {msg}")
             except Exception as e:
                 logger.warning(f"Receive loop ended: {e}")
 

python/src/mesh_lib/abstract.py

@@ -1,14 +1,14 @@
 from abc import ABC, abstractmethod
 from typing import Callable, Awaitable, List, Optional, Any, TypedDict, Union
-from mesh_lib.shared.types import Message, Node
+from mesh_lib.shared.types import Node
 
 class IServiceHandlers(TypedDict, total=False):
     on_request_channel: Optional[Callable[['IChannel'], Awaitable[None]]]
-    on_request_reply: Optional[Callable[[Message], Awaitable[Message]]]
+    on_request_reply: Optional[Callable[[Any], Awaitable[Any]]]
 
 class IServiceClient(ABC):
     @abstractmethod
-    async def request_reply(self, session_id: str, message: Message) -> Message:
+    async def request_reply(self, session_id: str, message: Any) -> Any:
         pass
 
     @abstractmethod
@@ -68,14 +68,14 @@ async def close(self):
 
 class IChannel(ABC):
     @abstractmethod
-    async def send(self, message: Message):
+    async def send(self, message: Any):
         """
         Send a message over this channel.
         """
         pass
 
     @abstractmethod
-    async def receive(self) -> Message:
+    async def receive(self) -> Any:
         """
         Receive the next message from this channel.
         """

python/src/mesh_lib/communication/h2_communication.py

@@ -1,7 +1,7 @@
 import asyncio
 import json
-from typing import Callable, Awaitable, Dict, List, Union
-from mesh_lib.shared.types import Message, MeshConfig, Node
+from typing import Callable, Awaitable, Dict, List, Union, Any
+from mesh_lib.shared.types import MeshConfig, Node
 from mesh_lib.communication.h2_protocol import H2Protocol
 from mesh_lib.abstract import ICommunication, IChannel, IServiceDiscovery
 
@@ -12,16 +12,13 @@ def __init__(self, protocol: H2Protocol, stream_id: int):
         self.protocol.ensure_stream_queue(stream_id)
         self.buffer = b''
 
-    async def send(self, message: Message):
-        payload = json.dumps({
-            "function_name": message.function_name,
-            "payload": message.payload
-        }).encode('utf-8')
+    async def send(self, message: Any):
+        payload = json.dumps(message).encode('utf-8')
         length = len(payload)
         header = length.to_bytes(4, byteorder='big')
         self.protocol.send_data(self.stream_id, header + payload)
 
-    async def receive(self) -> Message:
+    async def receive(self) -> Any:
         # Read until we have at least 4 bytes for length
         while len(self.buffer) < 4:
             chunk = await self.protocol.read_stream(self.stream_id)
@@ -45,7 +42,7 @@ async def receive(self) -> Message:
 
         try:
             obj = json.loads(payload_data.decode('utf-8'))
-            return Message(**obj)
+            return obj
         except Exception as e:
             print(f"Decode error: {e}")
             raise
@@ -73,7 +70,7 @@ 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[[Message], Awaitable[Message]]] = {}
+        self.unary_listeners: Dict[str, Callable[[Any], Awaitable[Any]]] = {}
         self._server = None
 
     async def start(self):
@@ -145,7 +142,7 @@ def _handle_channel(self, stream_id: int, service_name: str, protocol: H2Protoco
     def listen_channel(self, service_name: str, callback: Callable[[IChannel], Awaitable[None]]):
         self.listeners[service_name] = callback
 
-    def listen_unary(self, service_name: str, callback: Callable[[Message], Awaitable[Message]]):
+    def listen_unary(self, service_name: str, callback: Callable[[Any], Awaitable[Any]]):
         self.unary_listeners[service_name] = callback
 
     async def create_channel(self, target: Union[str, Node]) -> IChannel:
@@ -184,7 +181,7 @@ async def create_channel(self, target: Union[str, Node]) -> IChannel:
 
         return H2Channel(protocol, stream_id)
 
-    async def send_unary(self, node: Node, message: Message) -> Message:
+    async def send_unary(self, node: Node, message: Any) -> Any:
         loop = asyncio.get_running_loop()
         try:
              transport, protocol = await loop.create_connection(

python/src/mesh_lib/mesh.py

@@ -2,7 +2,7 @@
 import uuid
 import socket
 from typing import Optional, List, Callable, Awaitable
-from mesh_lib.shared.types import MeshConfig, Node, Message
+from mesh_lib.shared.types import MeshConfig, Node
 from mesh_lib.service_discovery import RedisServiceDiscovery
 from mesh_lib.communication import H2Communication
 from mesh_lib.communication import H2Communication