chore: add workerstop

Author: loks0n

src/Queue/Adapter/Swoole.php

@@ -3,8 +3,9 @@
 namespace Utopia\Queue\Adapter;
 
 use Swoole\Constant;
+use Swoole\Process;
 use Swoole\Process\Pool;
-use Utopia\CLI\Console;
+use Utopia\Console;
 use Utopia\Queue\Adapter;
 use Utopia\Queue\Consumer;
 
@@ -15,8 +16,12 @@ class Swoole extends Adapter
     /** @var callable */
     private $onStop;
 
-    public function __construct(Consumer $consumer, int $workerNum, string $queue, string $namespace = 'utopia-queue')
-    {
+    public function __construct(
+        Consumer $consumer,
+        int $workerNum,
+        string $queue,
+        string $namespace = "utopia-queue",
+    ) {
         parent::__construct($workerNum, $queue, $namespace);
 
         $this->consumer = $consumer;
@@ -25,25 +30,22 @@ public function __construct(Consumer $consumer, int $workerNum, string $queue, s
 
     public function start(): self
     {
-        $this->pool->set(['enable_coroutine' => true]);
-
-        // Register signal handlers in the main process before starting pool
-        if (extension_loaded('pcntl')) {
-            pcntl_signal(SIGTERM, function () {
-                Console::info("[Swoole] Received SIGTERM, initiating graceful shutdown...");
-                $this->stop();
-            });
-
-            pcntl_signal(SIGINT, function () {
-                Console::info("[Swoole] Received SIGINT, initiating graceful shutdown...");
-                $this->stop();
-            });
+        $this->pool->set(["enable_coroutine" => true]);
+
+        // Register signal handlers
+        Process::signal(SIGTERM, function () {
+            Console::info(
+                "[Swoole] Received SIGTERM, initiating graceful shutdown...",
+            );
+            $this->stop();
+        });
 
-            // Enable async signals
-            pcntl_async_signals(true);
-        } else {
-            Console::warning("[Swoole] pcntl extension is not loaded, worker will not shutdown gracefully.");
-        }
+        Process::signal(SIGINT, function () {
+            Console::info(
+                "[Swoole] Received SIGINT, initiating graceful shutdown...",
+            );
+            $this->stop();
+        });
 
         $this->pool->start();
         return $this;
@@ -52,7 +54,7 @@ public function start(): self
     public function stop(): self
     {
         if ($this->onStop) {
-            call_user_func($this->onStop);
+            \call_user_func($this->onStop);
         }
 
         Console::info("[Swoole] Shutting down process pool...");
@@ -63,23 +65,26 @@ public function stop(): self
 
     public function workerStart(callable $callback): self
     {
-        $this->pool->on(Constant::EVENT_WORKER_START, function (Pool $pool, string $workerId) use ($callback) {
-            // Register signal handlers in each worker process for graceful shutdown
-            if (extension_loaded('pcntl')) {
-                pcntl_signal(SIGTERM, function () use ($workerId) {
-                    Console::info("[Worker] Worker {$workerId} received SIGTERM, closing consumer...");
-                    $this->consumer->close();
-                });
-
-                pcntl_signal(SIGINT, function () use ($workerId) {
-                    Console::info("[Worker] Worker {$workerId} received SIGINT, closing consumer...");
-                    $this->consumer->close();
-                });
-
-                pcntl_async_signals(true);
-            }
-
-            call_user_func($callback, $workerId);
+        $this->pool->on(Constant::EVENT_WORKER_START, function (
+            Pool $pool,
+            string $workerId,
+        ) use ($callback) {
+            // Register signal handlers in worker
+            Process::signal(SIGTERM, function () {
+                Console::info(
+                    "[Swoole] Received SIGTERM, initiating graceful shutdown...",
+                );
+                $this->stop();
+            });
+
+            Process::signal(SIGINT, function () {
+                Console::info(
+                    "[Swoole] Received SIGINT, initiating graceful shutdown...",
+                );
+                $this->stop();
+            });
+
+            \call_user_func($callback, $workerId);
         });
 
         return $this;
@@ -88,8 +93,11 @@ public function workerStart(callable $callback): self
     public function workerStop(callable $callback): self
     {
         $this->onStop = $callback;
-        $this->pool->on(Constant::EVENT_WORKER_STOP, function (Pool $pool, string $workerId) use ($callback) {
-            call_user_func($callback, $workerId);
+        $this->pool->on(Constant::EVENT_WORKER_STOP, function (
+            Pool $pool,
+            string $workerId,
+        ) use ($callback) {
+            \call_user_func($callback, $workerId);
         });
 
         return $this;

src/Queue/Broker/Pool.php

@@ -12,8 +12,7 @@
     public function __construct(
         private ?UtopiaPool $publisher = null,
         private ?UtopiaPool $consumer = null,
-    ) {
-    }
+    ) {}
 
     public function enqueue(Queue $queue, array $payload): bool
     {
@@ -30,8 +29,12 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
         return $this->delegatePublish(__FUNCTION__, \func_get_args());
     }
 
-    public function consume(Queue $queue, callable $messageCallback, callable $successCallback, callable $errorCallback): void
-    {
+    public function consume(
+        Queue $queue,
+        callable $messageCallback,
+        callable $successCallback,
+        callable $errorCallback,
+    ): void {
         $this->delegateConsumer(__FUNCTION__, \func_get_args());
     }
 
@@ -42,14 +45,20 @@ public function close(): void
 
     protected function delegatePublish(string $method, array $args): mixed
     {
-        return $this->publisher?->use(function (Publisher $adapter) use ($method, $args) {
+        return $this->publisher?->use(function (Publisher $adapter) use (
+            $method,
+            $args,
+        ) {
             return $adapter->$method(...$args);
         });
     }
 
     protected function delegateConsumer(string $method, array $args): mixed
     {
-        return $this->consumer?->use(function (Consumer $adapter) use ($method, $args) {
+        return $this->consumer?->use(function (Consumer $adapter) use (
+            $method,
+            $args,
+        ) {
             return $adapter->$method(...$args);
         });
     }