eee

Processes/AbstractProcess.php

@@ -133,7 +133,7 @@ abstract class AbstractProcess implements OnProcessInterface
             Coroutine::set($array);
             Coroutine::create(fn() => $this->coroutineWaitSignal());
         } else {
-            pcntl_signal(SIGTERM | SIGINT | SIGUSR1, [$this, 'pointWaitSignal']);
+            $process::signal(SIGTERM | SIGINT | SIGUSR1, [$this, 'pointWaitSignal']);
         }
         return $this;
     }

Task/OnTaskFinish.php

@@ -6,12 +6,12 @@ class OnTaskFinish
 {
 
 
-	/**
-	 * @param int $task_id
-	 * @param mixed $data
-	 */
-	public function __construct(int $task_id, mixed $data)
-	{
-	}
+    /**
+     * @param int $task_id
+     * @param mixed $data
+     */
+    public function __construct(public int $task_id, public mixed $data)
+    {
+    }
 
 }

Task/OnTaskInterface.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace Kiri\Server\Task;
+
+interface OnTaskInterface
+{
+
+
+    /**
+     * @param int $task_id
+     * @param int $src_worker_id
+     * @return mixed
+     */
+    public function process(int $task_id, int $src_worker_id): mixed;
+
+
+}

Task/Task.php

@@ -67,22 +67,24 @@ 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
     {
         try {
-            $data = json_decode($data, true);
-            if (is_null($data)) {
-                return null;
-            }
             [$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);
         }
     }
 

Task/TaskExecute.php

@@ -2,22 +2,50 @@
 
 namespace Kiri\Server\Task;
 
+use Exception;
 use Kiri\Server\ServerInterface;
 
 class TaskExecute implements TaskInterface
 {
 
+
     /**
+     * @param string $handler
+     * @param mixed $data
+     * @param int $dstWorkerId
+     * @param callable|null $finishFinishCallback
+     * @return void
+     * @throws
+     */
+    public function task(string $handler, mixed $data, int $dstWorkerId = -1, ?callable $finishFinishCallback = null): void
+    {
+        $array = class_implements($handler, true);
+        if (!in_array(OnTaskInterface::class, $array, true)) {
+            throw new Exception('Task is not implement OnTaskInterface');
+        }
+        $server = \Kiri::getDi()->get(ServerInterface::class);
+
+        $server->task([$handler, $data], $dstWorkerId, $finishFinishCallback);
+    }
+
+
+    /**
+     * @param string $handler
      * @param mixed $data
      * @param float $timeout
      * @param int $dstWorkerId
      * @return mixed
+     * @throws
      */
-    public function taskWait(mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed
+    public function taskWait(string $handler, mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed
     {
+        $array = class_implements($handler, true);
+        if (!in_array(OnTaskInterface::class, $array, true)) {
+            throw new Exception('Task is not implement OnTaskInterface');
+        }
         $server = \Kiri::getDi()->get(ServerInterface::class);
 
-        return $server->taskwait($data, $timeout, $dstWorkerId);
+        return $server->taskwait([$handler, $data], $timeout, $dstWorkerId);
     }
 
 
@@ -45,6 +73,4 @@ class TaskExecute implements TaskInterface
 
         return $server->taskWaitMulti($tasks, $timeout);
     }
-
-
-}
+}

Task/TaskInterface.php

@@ -7,11 +7,24 @@ interface TaskInterface
 
 
     /**
-     * @param array $tasks
-     * @param float $timeout
-     * @return false|array
+     * @param string $handler
+     * @param mixed $data
+     * @param int $dstWorkerId
+     * @param callable|null $finishFinishCallback
+     * @return void
      */
-    public function taskWaitMulti(array $tasks, float $timeout = 0.5): false|array;
+    public function task(string $handler, mixed $data, int $dstWorkerId = -1, ?callable $finishFinishCallback = null): void;
+
+
+    /**
+     * @param string $handler
+     * @param mixed $data
+     * @param float $timeout
+     * @param int $dstWorkerId
+     * @return mixed
+     */
+    public function taskWait(string $handler, mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed;
+
 
     /**
      * @param array $tasks
@@ -20,12 +33,12 @@ interface TaskInterface
      */
     public function taskCo(array $tasks, float $timeout = 0.5): false|array;
 
+
     /**
-     * @param mixed $data
+     * @param array $tasks
      * @param float $timeout
-     * @param int $dstWorkerId
-     * @return mixed
+     * @return false|array
      */
-    public function taskWait(mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed;
+    public function taskWaitMulti(array $tasks, float $timeout = 0.5): false|array;
 
 }

Task/TestTask.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace Kiri\Server\Task;
+
+class TestTask implements OnTaskInterface
+{
+
+
+    public function __construct($user, $friend)
+    {
+    }
+
+
+    /**
+     * @param int $task_id
+     * @param int $src_worker_id
+     * @return int
+     */
+    public function process(int $task_id, int $src_worker_id): int
+    {
+        return 0;
+    }
+
+}