eee

Abstracts/AsyncServer.php

@@ -92,7 +92,7 @@ class AsyncServer extends Component implements ServerInterface
             $config->events[Constant::SHUTDOWN] = [OnServer::class, 'onShutdown'];
         }
 
-        $this->event($this->server, array_merge(\config('server.events', []), $config->events));
+        $this->event($this->server, array_merge(\config('servers.server.events', []), $config->events));
         $this->container->bind(ServerInterface::class, $this->server);
         return $service;
     }
@@ -106,7 +106,7 @@ class AsyncServer extends Component implements ServerInterface
      */
     protected function systemConfig(SConfig $config, int $daemon): array
     {
-        $settings                                     = array_merge(\config('server.settings', []), $config->settings);
+        $settings                                     = array_merge(\config('servers.server.settings', []), $config->settings);
         $settings[Constant::OPTION_DAEMONIZE]         = (bool)$daemon;
         $settings[Constant::OPTION_ENABLE_REUSE_PORT] = true;
         $settings[Constant::OPTION_PID_FILE]          = storage('.swoole.pid');

Abstracts/HotReload.php

@@ -113,7 +113,7 @@ class HotReload extends AbstractProcess
         }
         $this->reloading = true;
 
-        di(StdoutLogger::class)->println('reloading server[' . \config('id', 'system-service') . '], please waite.');
+        di(StdoutLogger::class)->println('reloading server[' . \config('site.id', 'system-service') . '], please waite.');
 
         $this->clear();
         di(ServerInterface::class)->reload();
@@ -128,7 +128,7 @@ class HotReload extends AbstractProcess
      */
     protected function addListen(): void
     {
-        foreach (config('reload.listen') as $value) {
+        foreach (config('servers.reload.listen') as $value) {
             $this->readDirectory($value);
         }
     }
@@ -168,4 +168,4 @@ class HotReload extends AbstractProcess
             }
         }
     }
-}
+}

Abstracts/Server.php

@@ -4,6 +4,8 @@
 namespace Kiri\Server\Abstracts;
 
 
+use Kiri\Error\StdoutLogger;
+
 /**
  * Class Server
  * @package Server\Abstracts
@@ -11,12 +13,22 @@ namespace Kiri\Server\Abstracts;
 abstract class Server
 {
 
-    /**
+	/**
-     * Server constructor.
+	 * Server constructor.
-     * @throws
+	 * @throws
-     */
+	 */
-    public function __construct()
+	public function __construct()
-    {
+	{
-    }
+	}
+
+
+
+	/**
+	 * @return StdoutLogger
+	 */
+	protected function getLogger(): StdoutLogger
+	{
+		return \Kiri::getLogger();
+	}
 
 }

Abstracts/TraitServer.php

@@ -21,7 +21,7 @@ trait TraitServer
      */
     public function onSignal(): void
     {
-        $signal = \config('signal', []);
+        $signal = \config('servers.signal', []);
         $this->onPcntlSignal(SIGINT, [$this, 'onSigint']);
         foreach ($signal as $sig => $value) {
             if (is_array($value) && is_string($value[0])) {
@@ -46,7 +46,7 @@ trait TraitServer
             Kiri::getLogger()->alert('Pid ' . getmypid() . ' get signo ' . $no);
             $this->shutdown();
         } catch (\Throwable $exception) {
-            error($exception);
+			\Kiri::getLogger()->json_log($exception);
         }
     }
 

Handler/OnPipeMessage.php

@@ -30,7 +30,7 @@ class OnPipeMessage extends Server
             }
             call_user_func([$message, 'process'], $server, $src_worker_id);
         } catch (\Throwable $throwable) {
-            Kiri::getLogger()->error(throwable($throwable));
+			\Kiri::getLogger()->json_log($throwable, ['src_worker_id' => $src_worker_id, 'message' => $message]);
         }
     }
 

Handler/OnServerWorker.php

@@ -30,135 +30,139 @@ class OnServerWorker extends Kiri\Server\Abstracts\Server
 {
 
 
-    /**
+	/**
-     * @var EventDispatch
+	 * @var EventDispatch
-     */
+	 */
-    #[Container(EventDispatch::class)]
+	#[Container(EventDispatch::class)]
-    public EventDispatch $dispatch;
+	public EventDispatch $dispatch;
 
 
-    /**
+	/**
-     * @return void
+	 * @return void
-     */
+	 */
-    public function init(): void
+	public function init(): void
-    {
+	{
-        on(OnBeforeWorkerStart::class, [$this, 'onWorkerNameAlias']);
+		on(OnBeforeWorkerStart::class, [$this, 'onWorkerNameAlias']);
-    }
+	}
 
 
-    /**
+	/**
-     * @param OnBeforeWorkerStart $workerStart
+	 * @param OnBeforeWorkerStart $workerStart
-     * @return void
+	 * @return void
-     */
+	 */
-    public function onWorkerNameAlias(OnBeforeWorkerStart $workerStart): void
+	public function onWorkerNameAlias(OnBeforeWorkerStart $workerStart): void
-    {
+	{
-        set_env('environmental_workerId', $workerStart->workerId);
+		if ($workerStart->workerId < $workerStart->server->setting['worker_num']) {
-        if ($workerStart->workerId < $workerStart->server->setting['worker_num']) {
+			$this->processName($workerStart->server, 'Worker');
-            $this->processName($workerStart->server, 'Worker');
+			set_env('environmental', Kiri::WORKER);
-            set_env('environmental', Kiri::WORKER);
+		} else {
-        } else {
+			$this->processName($workerStart->server, 'Tasker');
-            $this->processName($workerStart->server, 'Tasker');
+			set_env('environmental', Kiri::TASK);
-            set_env('environmental', Kiri::TASK);
+		}
-        }
+		set_env('environmental_worker_id', $workerStart->workerId);
-    }
+	}
 
 
-    /**
+	/**
-     * @param Server $server
+	 * @param Server $server
-     * @param int $workerId
+	 * @param int $workerId
-     * @return void
+	 * @return void
-     * @throws
+	 * @throws
-     */
+	 */
-    public function onWorkerStart(Server $server, int $workerId): void
+	public function onWorkerStart(Server $server, int $workerId): void
-    {
+	{
-        $this->dispatch->dispatch(new OnBeforeWorkerStart($server, $workerId));
+		try {
-        if ($workerId < $server->setting['worker_num']) {
+			$this->dispatch->dispatch(new OnBeforeWorkerStart($server, $workerId));
-            CoordinatorManager::utility(Coordinator::WORKER_START)->waite();
+			if ($workerId < $server->setting['worker_num']) {
+				CoordinatorManager::utility(Coordinator::WORKER_START)->waite();
 
-            $this->dispatch->dispatch(new OnWorkerStart($server, $workerId));
+				$this->dispatch->dispatch(new OnWorkerStart($server, $workerId));
-        } else {
+			} else {
-            $this->dispatch->dispatch(new OnTaskerStart($server, $workerId));
+				$this->dispatch->dispatch(new OnTaskerStart($server, $workerId));
-        }
+			}
-        $this->dispatch->dispatch(new OnAfterWorkerStart($server, $workerId));
+		} catch (Throwable $exception) {
-    }
+			\Kiri::getLogger()->json_log($exception);
+		} finally {
+			$this->dispatch->dispatch(new OnAfterWorkerStart($server, $workerId));
+		}
+	}
 
 
-    /**
+	/**
-     * @param Server $server
+	 * @param Server $server
-     * @param string $prefix
+	 * @param string $prefix
-     * @return void
+	 * @return void
-     */
+	 */
-    protected function processName(Server $server, string $prefix): void
+	protected function processName(Server $server, string $prefix): void
-    {
+	{
-        Kiri::setProcessName(sprintf($prefix . '[%d]', $server->worker_pid));
+		Kiri::setProcessName(sprintf($prefix . '[%d]', $server->worker_pid));
-    }
+	}
 
 
-    /**
+	/**
-     * @param Server $server
+	 * @param Server $server
-     * @param int $workerId
+	 * @param int $workerId
-     * @throws
+	 * @throws
-     */
+	 */
-    public function onWorkerStop(Server $server, int $workerId): void
+	public function onWorkerStop(Server $server, int $workerId): void
-    {
+	{
-        event(new OnWorkerStop($server, $workerId));
+		event(new OnWorkerStop($server, $workerId));
-        Timer::clearAll();
+		Timer::clearAll();
-    }
+	}
 
 
-    /**
+	/**
-     * @param Server $server
+	 * @param Server $server
-     * @param int $workerId
+	 * @param int $workerId
-     * @throws
+	 * @throws
-     */
+	 */
-    public function onWorkerExit(Server $server, int $workerId): void
+	public function onWorkerExit(Server $server, int $workerId): void
-    {
+	{
-        event(new OnWorkerExit($server, $workerId));
+		event(new OnWorkerExit($server, $workerId));
-    }
+	}
 
 
-    /**
+	/**
-     * @param Server $server
+	 * @param Server $server
-     * @param int $worker_id
+	 * @param int $worker_id
-     * @param int $worker_pid
+	 * @param int $worker_pid
-     * @param int $exit_code
+	 * @param int $exit_code
-     * @param int $signal
+	 * @param int $signal
-     * @throws
+	 * @throws
-     */
+	 */
-    public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal): void
+	public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal): void
-    {
+	{
-        event(new OnWorkerError($server, $worker_id, $worker_pid, $exit_code, $signal));
+		event(new OnWorkerError($server, $worker_id, $worker_pid, $exit_code, $signal));
 
-        debug_print_backtrace(DEBUG_BACKTRACE_PROVIDE_OBJECT);
+		debug_print_backtrace(DEBUG_BACKTRACE_PROVIDE_OBJECT);
 
-        /** @var RequestInterface $context */
+		/** @var RequestInterface $context */
-        $context = Kiri\Di\Context::get(RequestInterface::class);
+		$context = Kiri\Di\Context::get(RequestInterface::class);
-        if (is_null($context)) {
+		if (is_null($context)) {
-            $message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal));
+			$message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal));
-        } else {
+		} else {
-            $message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s, method: %s, path: %s, query: %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal),
+			$message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s, method: %s, path: %s, query: %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal),
-                $context->getMethod(), $context->getUri()->getPath(), $context->getUri()->getQuery());
+							   $context->getMethod(), $context->getUri()->getPath(), $context->getUri()->getQuery());
-        }
+		}
-        error($message . PHP_EOL);
+		$this->getLogger()->println($message);
-
+		$this->system_mail($message);
-        $this->system_mail($message);
+	}
-    }
 
 
-    /**
+	/**
-     * @param $messageContent
+	 * @param $messageContent
-     * @throws
+	 * @throws
-     */
+	 */
-    protected function system_mail($messageContent): void
+	protected function system_mail($messageContent): void
-    {
+	{
-        try {
+		try {
-            $email = config('email', ['enable' => false]);
+			$email = config('email', ['enable' => false]);
-            if (!empty($email) && ($email['enable'] ?? false)) {
+			if (!empty($email) && ($email['enable'] ?? false)) {
-                Help::sendEmail($email, 'Service Error', $messageContent);
+				Help::sendEmail($email, 'Service Error', $messageContent);
-            }
+			}
-        } catch (Throwable $e) {
+		} catch (Throwable $e) {
-            error($e, ['email']);
+			\Kiri::getLogger()->json_log($e);
-        }
+		}
-    }
+	}
 
 }

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;
     }

Processes/TraitProcess.php

@@ -40,7 +40,7 @@ trait TraitProcess
     private function genProcess(AbstractProcess $name): Process
     {
         return new Process(function (Process $process) use ($name) {
-            $process->name('[' . \config('id', 'system-service') . '].' . $name->getName() . '[' . $process->pid . ']');
+            $process->name('[' . \config('site.id', 'system-service') . '].' . $name->getName() . '[' . $process->pid . ']');
             $name->onShutdown($process)->process($process);
         },
             $name->getRedirectStdinAndStdout(),
@@ -66,4 +66,4 @@ trait TraitProcess
     {
         return $this->_process;
     }
-}
+}

ServerCommand.php

@@ -33,6 +33,7 @@ class ServerCommand extends Command
 {
 
 
+    #[Container(State::class)]
     public State $state;
 
 
@@ -64,17 +65,6 @@ class ServerCommand extends Command
     public EventProvider $eventProvider;
 
 
-    /**
-     * @param string|null $name
-     * @throws Exception
-     */
-    public function __construct(string $name = null)
-    {
-        parent::__construct($name);
-        $this->state = Kiri::getDi()->get(State::class);
-    }
-
-
     /**
      * @return void
      */
@@ -124,7 +114,7 @@ class ServerCommand extends Command
      */
     protected function stop(): int
     {
-        $configs   = config('server', []);
+        $configs   = config('servers.server', []);
         $instances = $this->asyncServer->sortService($configs['ports'] ?? []);
         foreach ($instances as $config) {
             $this->state->exit($config->port);
@@ -141,13 +131,13 @@ class ServerCommand extends Command
      */
     protected function start(InputInterface $input): int
     {
-        $this->asyncServer->addProcess(config('processes', []));
+        $this->asyncServer->addProcess(config('process', []));
-        if (\config('reload.hot', false) === true) {
+        if (\config('servers.reload.hot', false) === true) {
             $this->asyncServer->addProcess([HotReload::class]);
         } else {
             di(Router::class)->scan_build_route();
         }
-        $this->asyncServer->initCoreServers(config('server', []), (int)$input->getOption('daemon'));
+        $this->asyncServer->initCoreServers(config('servers.server', []), (int)$input->getOption('daemon'));
         $this->asyncServer->start();
         return 1;
     }

ServerProviders.php

@@ -23,6 +23,6 @@ class ServerProviders extends Providers
         $server = $this->container->get(ServerCommand::class);
 
         $console = $this->container->get(Application::class);
-        $console->add($server);
+        $console->addCommand($server);
     }
 }

State.php

@@ -21,7 +21,7 @@ class State extends Component
      */
     public function init(): void
     {
-        $this->servers = config('server.ports');
+        $this->servers = config('servers.server.ports');
     }
 
 

Task/OnTaskFinish.php

@@ -6,12 +6,12 @@ class OnTaskFinish
 {
 
 
-	/**
+    /**
-	 * @param int $task_id
+     * @param int $task_id
-	 * @param mixed $data
+     * @param mixed $data
-	 */
+     */
-	public function __construct(int $task_id, 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

@@ -24,7 +24,7 @@ class Task
      */
     public function __construct()
     {
-        $exception = \config('exception.task');
+        $exception = \config('servers.task.exception');
         if (!in_array(ExceptionHandlerInterface::class, class_implements($exception))) {
             $exception = ExceptionHandlerDispatcher::class;
         }
@@ -67,22 +67,26 @@ 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]);
+            $handler = Kiri::getDi()->make($handler, $params);
-            return call_user_func($handler, $task_id, $src_worker_id, $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());
+			\Kiri::getLogger()->json_log($throwable, ['task_id' => $task_id, 'src_worker_id' => $src_worker_id, 'data' => $data]);
+
+			$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 string $handler
-     * @param float $timeout
+     * @param mixed $data
-     * @return false|array
+     * @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 false|array
-     * @return mixed
      */
-    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;
+    }
+
+}

composer.json

@@ -9,7 +9,7 @@
     }
   ],
   "require": {
-    "php": ">=8.0",
+    "php": ">=8.4",
     "ext-json": "*",
     "composer-runtime-api": "^2.0",
     "psr/http-server-middleware": "^1.0",