revert to use async method onData

Author: randilfernando

node/README.md

@@ -46,7 +46,7 @@ mesh.registerService('my-service-name')
     .onRequestChannel(async (channel) => {
         console.log(`Accepted connection`);
         try {
-            channel.onData(async (msg) => {
+            await channel.onData(async (msg) => {
                 console.log('Received:', msg.payload);
                 await channel.send({ function_name: 'reply', payload: 'Got it!' });
             });
@@ -73,7 +73,7 @@ await channel.send({
 });
 
 // Reading messages (Callback handles backpressure)
-channel.onData(async (msg) => {
+await channel.onData(async (msg) => {
     console.log('Reply:', msg.payload);
     // To stop receiving, you can close the channel from within handler or check explicitly
     if (msg.payload === 'Hello World') {
@@ -92,11 +92,11 @@ await mesh.close();
 ```
 
 ### 6. Callback-based Backpressure
-Mesh supports `onData(handler)` for consumers who prefer callbacks but need backpressure. The method returns `void` (fire-and-forget), and the processing loop runs in the background.
+Mesh supports `onData(handler)` for consumers who prefer callbacks but need backpressure. The method returns a `Promise` that resolves when the stream is closed, allowing you to `await` the entire processing session.
 
 ```typescript
 ```typescript
-channel.onData(async (msg) => {
+await channel.onData(async (msg) => {
     await processMessage(msg);
     // The next message is only requested after this line completes
 });

node/mesh_peer.ts

@@ -56,7 +56,7 @@ async function main() {
 
                 // Receive Loop
                 const receiveLoop = async () => {
-                    channel.onData(async (msg) => {
+                    await channel.onData(async (msg) => {
                         console.log(`[Node] Received: ${msg.payload}`);
                     });
                 };

node/src/abstract.ts

@@ -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;
-    onData(handler: (msg: any) => Promise<void>): void;
+    onData(handler: (msg: any) => Promise<void>): Promise<void>;
     [Symbol.asyncIterator](): AsyncIterator<any>;
 }
 

node/src/communication/h2-communication.ts

@@ -122,12 +122,10 @@ export class H2Channel implements IChannel {
         }
     }
 
-    public onData(handler: (msg: any) => Promise<void>): void {
-        (async () => {
-            for await (const msg of this) {
-                await handler(msg);
-            }
-        })();
+    public async onData(handler: (msg: any) => Promise<void>): Promise<void> {
+        for await (const msg of this) {
+            await handler(msg);
+        }
     }
 
     public async *[Symbol.asyncIterator](): AsyncIterator<any> {

python/README.md

@@ -59,7 +59,7 @@ async def stream_handler(channel):
             response = {"function_name": "reply", "payload": "Got it!"}
             await channel.send(response)
 
-        channel.on_data(handler)
+        await channel.on_data(handler)
     except Exception:
         pass
 
@@ -88,18 +88,18 @@ async def reply_handler(msg):
 async def reply_handler(msg):
     print(f"Reply: {msg['payload']}")
 
-channel.on_data(reply_handler)
+await channel.on_data(reply_handler)
 
 await channel.close()
 ```
 
 ### 5. Callback-based Backpressure
-Mesh supports `on_data(handler)` for consuming messages with backpressure. The method returns `None` (fire-and-forget), and the processing loop runs in the background using `asyncio.create_task`.
+Mesh supports `on_data(handler)` for consuming messages with backpressure. The method returns an `Awaitable` that resolves when the stream is closed, allowing you to `await` the entire processing session.
 
 ```python
 async def handler(msg):
     await process_message(msg)
     # The next message is consumed only after this function returns
 
-channel.on_data(handler)
+await channel.on_data(handler)
 ```

python/mesh_peer.py

@@ -50,7 +50,7 @@ async def receive_loop():
                 async def msg_handler(msg):
                     logger.info(f"Received: {msg}")
 
-                channel.on_data(msg_handler)
+                await channel.on_data(msg_handler)
             except Exception as e:
                 logger.warning(f"Receive loop ended: {e}")
 

python/src/mesh/abstract.py

@@ -95,11 +95,10 @@ def __aiter__(self):
         return self
 
     @abstractmethod
-    def on_data(self, handler: Callable[[Any], Awaitable[None]]) -> None:
+    async def on_data(self, handler: Callable[[Any], Awaitable[None]]):
         """
         Process messages with a callback for backpressure.
         The next message is read only after the handler completes.
-        Run in background.
         """
         pass
 

python/src/mesh/communication/h2_communication.py

@@ -64,11 +64,9 @@ async def __aiter__(self):
                     return
                 raise e
 
-    def on_data(self, handler: Callable[[Any], Awaitable[None]]) -> None:
-        async def loop():
-            async for msg in self:
-                await handler(msg)
-        asyncio.create_task(loop())
+    async def on_data(self, handler: Callable[[Any], Awaitable[None]]):
+        async for msg in self:
+            await handler(msg)
 
 
 class H2Communication(ICommunication):