Merge pull request #7 from utopia-php/feat-di-in-worker-start

injections in worker start
utopia-php · Oct 31, 2022 · 0cad4cf · 0cad4cf
2 parents 42b132c + 3fe9866
commit 0cad4cf
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 31 deletions.
diff --git a/README.md b/README.md
@@ -45,10 +45,12 @@ $server
     });
 
 $server
-    ->workerStart(function () {
+    ->workerStart()
+    ->action(function () {
         echo "Worker Started" . PHP_EOL;
-    })
-    ->start();
+    });
+
+$server->start();
 
 
 // Enqueue messages to the worker using the Redis adapter

diff --git a/src/Queue/Server.php b/src/Queue/Server.php
@@ -45,6 +45,13 @@ class Server
      */
     protected array $shutdownHooks = [];
 
+    /**
+     * Hook that is called when worker starts
+     *
+     * @var Hook
+     */
+    protected Hook $workerStartHook;
+
     /**
      * @var array
      */
@@ -178,28 +185,10 @@ public function init(): Hook
     public function start(): self
     {
         try {
-            $this->adapter->start();
-        } catch (Throwable $error) {
-            self::setResource('error', fn () => $error);
-            foreach ($this->errorHooks as $hook) {
-                call_user_func_array($hook->getAction(), $this->getArguments($hook));
-            }
-        }
-        return $this;
-    }
-
-    /**
-     * Is called when a Worker starts.
-     * @param callable $callback
-     * @return self
-     */
-    public function workerStart(callable $callback = null): self
-    {
-        try {
-            $this->adapter->workerStart(function (string $workerId) use ($callback) {
+            $this->adapter->workerStart(function (string $workerId) {
                 Console::success("[Worker] Worker {$workerId} is ready!");
-                if (!is_null($callback)) {
-                    call_user_func($callback);
+                if (!is_null($this->workerStartHook)) {
+                    call_user_func_array($this->workerStartHook->getAction(), $this->getArguments($this->workerStartHook));
                 }
                 while (true) {
                     /**
@@ -311,16 +300,28 @@ public function workerStart(callable $callback = null): self
                     }
                 }
             });
+
+            $this->adapter->start();
         } catch (Throwable $error) {
             self::setResource('error', fn () => $error);
             foreach ($this->errorHooks as $hook) {
                 call_user_func_array($hook->getAction(), $this->getArguments($hook));
             }
         }
-
         return $this;
     }
 
+    /**
+     * Is called when a Worker starts.
+     * @return Hook
+     */
+    public function workerStart(): Hook
+    {
+        $hook = new Hook();
+        $this->workerStartHook = $hook;
+        return $hook;
+    }
+
     /**
      * Is called when a Worker stops.
      * @param callable $callback

diff --git a/tests/Queue/servers/Swoole/worker.php b/tests/Queue/servers/Swoole/worker.php
@@ -24,7 +24,9 @@
     });
 
 $server
-    ->workerStart(function () {
+    ->workerStart()
+    ->action(function () {
         echo "Worker Started" . PHP_EOL;
-    })
-    ->start();
+    });
+
+$server->start();
diff --git a/tests/Queue/servers/Workerman/worker.php b/tests/Queue/servers/Workerman/worker.php
@@ -23,7 +23,9 @@
     });
 
 $server
-    ->workerStart(function () {
+    ->workerStart()
+    ->action(function () {
         echo "Worker Started" . PHP_EOL;
-    })
-    ->start();
+    });
+
+$server->start();