Refactor client handling in ChatServer to use async inherited scope for improved coroutine management and enhance connection cancellation in RabbitMQListener

Author: EdmondDantes

examples/ChatServer/ChatServer.php

@@ -22,43 +22,45 @@ function startChatServer(string $host, int $port): void
                 $clients[$id] = $client;
 
                 // Handle each client in a separate coroutine and child scope
-                spawn with Scope::inherit() use($client, $id, &$clients) {
-                    try {
-                        // Send welcome message
-                        socket_write($client, "Welcome to the chat room!\n");
-                        
-                        // Read messages from this client and broadcast them
-                        while (true) {
-                            $message = socket_read($client, 1024);
-                            
-                            if ($message === false || $message === '') {
-                                break; // Client disconnected
-                            }
-                            
-                            $broadcastMsg = "Client $id: $message";
-                            echo $broadcastMsg;
+                spawn use($client, $id, &$clients) {
+                    async inherited $clientScope {
+                        try {
+                            // Send welcome message
+                            socket_write($client, "Welcome to the chat room!\n");
 
-                            // Broadcast to all other clients
-                            foreach ($clients as $cid => $c) {
-                                if ($cid !== $id) {
-                                    spawn socket_write($c, $broadcastMsg);
+                            // Read messages from this client and broadcast them
+                            while (true) {
+                                $message = socket_read($client, 1024);
+                                
+                                if ($message === false || $message === '') {
+                                    break; // Client disconnected
+                                }
+                                
+                                $broadcastMsg = "Client $id: $message";
+                                echo $broadcastMsg;
+                                
+                                // Broadcast to all other clients
+                                foreach ($clients as $cid => $c) {
+                                    if ($cid !== $id) {
+                                        spawn socket_write($c, $broadcastMsg);
+                                    }
                                 }
                             }
-                        }
-                    } catch (Exception $e) {
-                        echo "Error handling client $id: " . $e->getMessage() . "\n";
-                    } finally {
-                        // Gracefully handle connection cancellation
-                        try {
-                            await $clientScope->allTask() until Async\timeout(2000);
+                        } catch (Exception $e) {
+                            echo "Error handling client $id: " . $e->getMessage() . "\n";
                         } finally {
-                            // Clean up when client disconnects
-                            socket_close($client);
-                            unset($clients[$id]);
-                            echo "Client $id disconnected\n";
+                            // Gracefully handle connection cancellation
+                            try {
+                                await $clientScope->allTask() until Async\timeout(2000);
+                            } finally {
+                                // Clean up when client disconnects
+                                socket_close($client);
+                                unset($clients[$id]);
+                                echo "Client $id disconnected\n";
+                            }
                         }
-                    }
-                };
+                    };
+                }
             }
         };
 

examples/Rabbitmq/RabbitMQListener.php

@@ -71,8 +71,11 @@ private function handleMessage(AMQPMessage $msg, string $queue): void
     public function stop(): void
     {
         $this->scope->cancel();
-        await($this->scope);
-        $this->connection->close();
+        try {
+            await $this->scope->allTasks();
+        } finally {
+            $this->connection->close();
+        }
     }
 
     public function cancelAll(): void