Refactor BackgroundLogger and ProcessPool to use TaskGroup for improved task management; update RabbitMQListener and WebScraper for consistency

Author: EdmondDantes

examples/BackgroundLogger/BackgroundLogger.php

@@ -5,19 +5,20 @@
 namespace BackgroundLogger;
 
 use Async\Scope;
+use Async\TaskGroup;
 
 class BackgroundLogger
 {
-    private Scope $scope;
+    private TaskGroup $taskGroup;
 
     public function __construct()
     {
-        $this->scope = new Scope();
+        $this->taskGroup = new TaskGroup(new Scope());
     }
 
     public function logAsync(string $message): void
     {
-        spawn with $this->scope static use($message) {
+        $this->taskGroup->spawn(static function() use($message) {
             try {
                 file_put_contents(
                     'app.log',
@@ -27,12 +28,12 @@ public function logAsync(string $message): void
             } catch (\Throwable $e) {
                 error_log("Async log failed: " . $e->getMessage());
             }
-        };
+        });
     }
 
     public function __destruct()
     {
-        $this->scope->dispose();
+        $this->taskGroup->dispose();
     }
 }
 

examples/ChatServer/ChatServer.php

@@ -51,7 +51,7 @@ function startChatServer(string $host, int $port): void
                         } finally {
                             // Gracefully handle connection cancellation
                             try {
-                                await $clientScope->allTask() until Async\timeout(2000);
+                                await $clientScope until Async\timeout(2000);
                             } finally {
                                 // Clean up when client disconnects
                                 socket_close($client);

examples/DistributedTaskOrchestration/DistributedTaskOrchestration.php

@@ -1,5 +1,7 @@
 <?php
 
+use function Async\currentContext;
+
 function orchestrateDistributedProcess(array $nodes, array $taskConfig): array
 {
     async $orchestratorScope {

examples/FanOut.php

@@ -13,17 +13,24 @@ function fetchUrl(string $url): string {
 function fetchAllUrls(array $urls): array
 {
     async bounded $scope {
+        
+        $tasks = new \Async\TaskGroup();
+        
         foreach ($urls as $url) {
-            spawn fetchUrl($url);
+            $tasks->add(spawn fetchUrl($url));
         }
 
         $results = [];
 
-        foreach (await $scope->directTasks() as $url => $future) {
-            $results[$url] = $future->getResult();
-        }
+        try {
+            foreach (await $tasks as $url => $future) {
+                $results[$url] = $future->getResult();
+            }
 
-        return $results;
+            return $results;
+        } finally {
+            $tasks->dispose();
+        }
     }
 }
 

examples/ProcessPool/ProcessPool.php

@@ -6,12 +6,13 @@
 
 use Async\CancellationException;
 use Async\Scope;
+use Async\TaskGroup;
 
 final class ProcessPool
 {
     private Scope $watcherScope;
-    private Scope $poolScope;
-    private Scope $jobsScope;
+    private TaskGroup $poolTasks;
+    private TaskGroup $jobsTasks;
     /**
      * List of pipes for each process.
      * @var array
@@ -27,32 +28,32 @@ final class ProcessPool
     public function __construct(readonly public string $entryPoint, readonly public int $max, readonly public int $min)
     {
         // Define the coroutine scopes for the pool, watcher, and jobs
-        $this->poolScope = new Scope();
+        $this->poolTasks    = new TaskGroup(new Scope());
+        $this->jobsTasks    = new TaskGroup(new Scope());
         $this->watcherScope = new Scope();
-        $this->jobsScope = new Scope();
     }
 
     public function __destruct()
     {
         $this->watcherScope->dispose();
-        $this->poolScope->dispose();
-        $this->jobsScope->dispose();
+        $this->poolTasks->dispose();
+        $this->jobsTasks->dispose();
     }
 
     public function start(): void
     {
         spawn with $this->watcherScope $this->processWatcher();
 
         for ($i = 0; $i < $this->min; $i++) {
-            spawn with $this->poolScope $this->startProcess();
+            $this->poolTasks->spawn($this->startProcess(...));
         }
     }
 
     public function stop(): void
     {
         $this->watcherScope->cancel();
-        $this->poolScope->cancel();
-        $this->jobsScope->cancel();
+        $this->poolTasks->dispose();
+        $this->jobsTasks->dispose();
     }
 
     /**
@@ -67,10 +68,10 @@ public function executeJob(mixed $job, callable $resultHandle): void
         $pid = array_search(true, $this->descriptors, true);
 
         if ($pid === false && count($this->descriptors) < $this->max) {
-            spawn with $this->poolScope $this->startProcess();
+            $this->poolTasks->spawn($this->startProcess(...));
 
             // Try to find a free process again after a short delay
-            spawn with $this->jobsScope use($job, $resultHandle) {
+            $this->jobsTasks->spawn(function() use($job, $resultHandle) {
                 usleep(100);
 
                 $pid = array_search(true, $this->descriptors, true);
@@ -80,14 +81,14 @@ public function executeJob(mixed $job, callable $resultHandle): void
                 }
 
                 $this->sendJobAndReceiveResult($pid, $job, $resultHandle);
-            };
+            });
 
             return;
         } elseif ($pid === false) {
             $resultHandle(new \Exception('No free process'));
         } else {
             $this->descriptors[$pid] = false;
-            spawn with $this->jobsScope $this->sendJobAndReceiveResult($pid, $job, $resultHandle);
+            $this->jobsTasks->spawn($this->sendJobAndReceiveResult(...), $pid, $job, $resultHandle);
         }
     }
 
@@ -96,12 +97,12 @@ private function processWatcher(): void
         while (true) {
 
             try {
-                await $this->poolScope->directTasks();
+                await $this->poolTasks;
             } catch (StopProcessException $exception)  {
                 echo "Process was stopped with message: {$exception->getMessage()}\n";
 
                 if($exception->getCode() !== 0 || count($this->descriptors) < $this->min) {
-                    spawn with $this->poolScope $this->startProcess();
+                    $this->poolTasks->spawn($this->startProcess(...));
                 }
             }
         }

examples/Rabbitmq/RabbitMQListener.php

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

examples/TaskGroup.php

@@ -6,8 +6,6 @@
 
 function taskGroup(array $files): void
 {
-    $scope = Scope::inherit();
-    
     $fileReadTask = function (string $fileName): string {
 
         if(!is_file($fileName)) {
@@ -27,17 +25,20 @@ function taskGroup(array $files): void
         return $result;
     };
 
-    $tasks = [];
+    $scope = Scope::inherit();
+    $tasks = new \Async\TaskGroup($scope);
 
     foreach ($files as $file) {
-        $tasks[$file] = spawn in $scope $fileReadTask($file);
+        $tasks->add(spawn in $scope $fileReadTask($file));
     }
 
     try {
-        foreach (await Async\all($tasks) as $file => $result) {
-            echo "File $file: $result\n";
+        foreach (await $tasks as $result) {
+            echo "File $result\n";
         }
     } catch (Exception $e) {
         echo "Caught exception: ", $e->getMessage();
+    } finally {
+        $tasks->dispose();
     }
 }

examples/WebScraper/WebScraper.php

@@ -39,7 +39,7 @@ function scrapeWebsites(array $urls, int $concurrency = 5): array
             };
         }
 
-        $scraperScope->awaitAll();
+        await $scraperScope;
         return $results;
     }
 }