eee

Task/Task.php

@@ -67,22 +67,28 @@ class Task
      * @param int $task_id
      * @param int $src_worker_id
      * @param mixed $data
-     * @return mixed
+     * @return void
      * @throws
      */
-    public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): mixed
+    public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): void
     {
+        $response = 'task data format failed.';
         try {
             $data = json_decode($data, true);
-            if (is_null($data)) {
-                return null;
-            }
-            [$handler, $params] = [$data[0], $data[1]];
+            if (!is_null($data)) {
+                [$handler, $params] = [$data[0], $data[1]];
 
-            $handler[0] = Kiri::getDi()->get($handler[0]);
-            return call_user_func($handler, $task_id, $src_worker_id, $params);
+                $handler = Kiri::getDi()->make($handler, $params);
+                if (!($handler instanceof OnTaskInterface)) {
+                    throw new \Exception('Task process must implements ' . OnTaskInterface::class);
+                }
+
+                $response = call_user_func([$handler, 'process'], $task_id, $src_worker_id);
+            }
         } catch (\Throwable $throwable) {
-            return $this->exception->emit($throwable, response());
+            $response = throwable($throwable);
+        } finally {
+            $server->finish($response);
         }
     }