added async iterator to read channel messages

Author: randilfernando

node/README.md

@@ -25,12 +25,12 @@ Define your Redis connection and service details.
 import { Mesh, MeshConfig } from '@marketrix/mesh';
 
 const config: MeshConfig = {
+    redis: { host: 'localhost', port: 6379 }, // Shared Redis Config
     service_discovery: {
-        redis_config: { host: 'localhost', port: 6379 },
         heartbeat_interval: 2,
         heartbeat_threshold: 3
     },
-    host: '127.0.0.1', // Your local IP/Host
+    // host: '127.0.0.1', // Optional: Auto-detected if omitted
     port: 0 // 0 for random available port
 };
 ```
@@ -51,14 +51,13 @@ await mesh.registerService('my-service-name', {
     // 1. Streaming Handler
     onRequestChannel: async (channel) => {
         console.log(`Accepted connection`);
-        while (true) {
-            try {
-                const msg = await channel.receive();
+        try {
+            for await (const msg of channel) {
                 console.log('Received:', msg.payload);
                 await channel.send({ function_name: 'reply', payload: 'Got it!' });
-            } catch (e) {
-                break;
             }
+        } catch (e) {
+            console.error(e);
         }
     },
     // 2. Unary Handler (Request/Reply)
@@ -73,14 +72,14 @@ await mesh.registerService('my-service-name', {
 Discover and connect to another service.
 
 ```typescript
-const channel = await mesh.requestChannel('target-service-name', 'session-id-123');
+const channel = await mesh.service('target-service-name').requestChannel('session-id-123');
 
 await channel.send({ 
     function_name: 'greet', 
     payload: 'Hello World' 
 });
 
-const reply = await channel.receive();
+const reply = await channel.receive(); // You can still use receive() for single messages
 console.log('Reply:', reply.payload);
 
 await channel.close();

node/mesh_peer.ts

@@ -57,8 +57,7 @@ async function main() {
 
                 // Receive Loop
                 const receiveLoop = async () => {
-                    while (true) {
-                        const msg = await channel.receive();
+                    for await (const msg of channel) {
                         console.log(`[Node] Received: ${msg.payload}`);
                     }
                 };

node/src/abstract.ts

@@ -14,6 +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>;
 }
 
 export interface ICommunication {

node/src/communication/h2-communication.ts

@@ -121,6 +121,20 @@ export class H2Channel implements IChannel {
             this.closeHandlers.push(handler);
         }
     }
+
+    public async *[Symbol.asyncIterator](): AsyncIterator<Message> {
+        while (!this.isClosed) {
+            try {
+                const msg = await this.receive();
+                yield msg;
+            } catch (e: any) {
+                if (e.message === 'Channel closed') {
+                    return;
+                }
+                throw e;
+            }
+        }
+    }
 }
 
 export class H2Communication extends EventEmitter implements ICommunication {

node/src/demo_client.ts

@@ -26,13 +26,13 @@ from mesh_lib import Mesh
 from mesh_lib.shared.types import MeshConfig, ServiceDiscoveryConfig, Message
 
 config = MeshConfig(
+    redis={"host": "localhost", "port": 6379}, # Shared Redis Config
     service_discovery=ServiceDiscoveryConfig(
-        redis_config={"host": "localhost", "port": 6379},
         heartbeat_interval=2,
         heartbeat_threshold=3
     ),
-    host="127.0.0.1", # Your local IP/Host
-    port=0 # 0 for random available port
+    # host="127.0.0.1", # Optional: Auto-detected if omitted
+    port=0
 )
 ```
 
@@ -60,14 +60,14 @@ Register a service name to handle incoming connections.
 ```python
 async def stream_handler(channel):
     print("Accepted connection")
-    while True:
-        try:
-            msg = await channel.receive()
+    print("Accepted connection")
+    try:
+        async for msg in channel:
             print(f"Received: {msg.payload}")
             response = Message(function_name="reply", payload="Got it!")
             await channel.send(response)
-        except Exception:
-            break
+    except Exception:
+        pass
 
 async def unary_handler(msg):
     print(f"Received Unary: {msg.payload}")
@@ -83,7 +83,7 @@ await mesh.register_service("my-service-name",
 Discover and connect to another service.
 
 ```python
-channel = await mesh.request_channel("target-service-name", "session-id-123")
+channel = await mesh.service("target-service-name").request_channel("session-id-123")
 
 out_msg = Message(function_name="greet", payload="Hello World")
 await channel.send(out_msg)

python/README.md

@@ -48,8 +48,7 @@ async def send_loop():
         # Start parallel receive loop
         async def receive_loop():
             try:
-                while True:
-                    msg = await channel.receive()
+                async for msg in channel:
                     logger.info(f"Received: {msg.payload}")
             except Exception as e:
                 logger.warning(f"Receive loop ended: {e}")

python/mesh_peer.py

@@ -88,6 +88,12 @@ async def close(self):
         """
         pass
 
+    def __aiter__(self):
+        """
+        Return an async iterator for receiving messages.
+        """
+        return self
+
 class ICommunication(ABC):
     @abstractmethod
     async def create_channel(self, target: Union[str, Node]) -> IChannel:

python/src/mesh_lib/abstract.py

@@ -54,6 +54,19 @@ async def close(self):
         # End stream
         self.protocol.send_data(self.stream_id, b'', end_stream=True)
 
+    async def __aiter__(self):
+        while True:
+            try:
+                msg = await self.receive()
+                yield msg
+            except Exception as e:
+                # Check for stream closed messages
+                # In receive(): raise Exception("Stream closed") or "Incomplete message... closed"
+                msg_str = str(e).lower()
+                if "stream closed" in msg_str or "incomplete" in msg_str:
+                    return
+                raise e
+
 
 class H2Communication(ICommunication):
     def __init__(self, config: MeshConfig, service_discovery: IServiceDiscovery):