first commit

Handler/OnPipeMessage.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace Server\Handler;
+
+use Annotation\Inject;
+use Server\Abstracts\Server;
+use Exception;
+use Server\SInterface\OnPipeMessageInterface;
+use Kiri\Events\EventDispatch;
+
+/**
+ *
+ */
+class OnPipeMessage extends Server
+{
+
+
+	/** @var EventDispatch  */
+	#[Inject(EventDispatch::class)]
+	public EventDispatch $eventDispatch;
+
+
+	/**
+	 * @param \Swoole\Server $server
+	 * @param int $src_worker_id
+	 * @param mixed $message
+	 * @throws Exception
+	 */
+	public function onPipeMessage(\Swoole\Server $server, int $src_worker_id, mixed $message)
+	{
+		if (!is_object($message) || !($message instanceof OnPipeMessageInterface)) {
+			return;
+		}
+		call_user_func([$message, 'process'], $server, $src_worker_id);
+	}
+
+
+}

Handler/OnServer.php

@@ -0,0 +1,58 @@
+<?php
+
+namespace Server\Handler;
+
+use Annotation\Inject;
+use Kiri\Events\EventDispatch;
+use Kiri\Exception\ConfigException;
+use Server\Abstracts\Server;
+use Server\Events\OnBeforeShutdown;
+use Server\Events\OnShutdown;
+use Server\Events\OnStart;
+
+
+/**
+ * Class OnServerDefault
+ * @package Server\Manager
+ */
+class OnServer extends Server
+{
+
+	/**
+	 * @var EventDispatch
+	 */
+	#[Inject(EventDispatch::class)]
+	public EventDispatch $eventDispatch;
+
+
+	/**
+	 * @param \Swoole\Server $server
+	 * @throws ConfigException
+	 */
+	public function onStart(\Swoole\Server $server)
+	{
+		$this->setProcessName(sprintf('start[%d].server', $server->master_pid));
+
+		$this->eventDispatch->dispatch(new OnStart($server));
+	}
+
+
+	/**
+	 * @param \Swoole\Server $server
+	 */
+	public function onBeforeShutdown(\Swoole\Server $server)
+	{
+		$this->eventDispatch->dispatch(new OnBeforeShutdown($server));
+	}
+
+
+	/**
+	 * @param \Swoole\Server $server
+	 */
+	public function onShutdown(\Swoole\Server $server)
+	{
+		$this->eventDispatch->dispatch(new OnShutdown($server));
+	}
+
+
+}

Handler/OnServerManager.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace Server\Handler;
+
+use Annotation\Inject;
+use Kiri\Events\EventDispatch;
+use Server\Abstracts\Server;
+use Kiri\Exception\ConfigException;
+use Server\Events\OnManagerStart;
+use Server\Events\OnManagerStop;
+
+
+/**
+ * Class OnServerManager
+ * @package Server\Manager
+ */
+class OnServerManager extends Server
+{
+
+	/**
+	 * @var EventDispatch
+	 */
+	#[Inject(EventDispatch::class)]
+	public EventDispatch $eventDispatch;
+
+
+	/**
+     * @param \Swoole\Server $server
+     * @throws ConfigException
+     */
+	public function onManagerStart(\Swoole\Server $server)
+	{
+        $this->setProcessName(sprintf('manger[%d].0', $server->manager_pid));
+
+		$this->eventDispatch->dispatch(new OnManagerStart($server));
+	}
+
+
+	/**
+	 * @param \Swoole\Server $server
+	 */
+	public function onManagerStop(\Swoole\Server $server)
+	{
+		$this->eventDispatch->dispatch(new OnManagerStop($server));
+	}
+
+
+}

Handler/OnServerReload.php

@@ -0,0 +1,43 @@
+<?php
+
+namespace Server\Handler;
+
+use Annotation\Inject;
+use Kiri\Events\EventDispatch;
+use Server\Events\OnAfterReload;
+use Server\Events\OnBeforeReload;
+use Swoole\Server;
+
+
+/**
+ *
+ */
+class OnServerReload
+{
+
+
+	/**
+	 * @var EventDispatch
+	 */
+	#[Inject(EventDispatch::class)]
+	public EventDispatch $eventDispatch;
+
+
+	/**
+	 * @param Server $server
+	 */
+	public function onBeforeReload(Server $server)
+	{
+		$this->eventDispatch->dispatch(new OnBeforeReload($server));
+	}
+
+
+	/**
+	 * @param Server $server
+	 */
+	public function onAfterReload(Server $server)
+	{
+		$this->eventDispatch->dispatch(new OnAfterReload($server));
+	}
+
+}

Handler/OnServerTask.php

@@ -0,0 +1,101 @@
+<?php
+
+
+namespace Server\Handler;
+
+
+use Annotation\Inject;
+use Kiri\Abstracts\Logger;
+use Kiri\Exception\ConfigException;
+use Kiri\Kiri;
+use ReflectionException;
+use Server\SInterface\OnTaskInterface;
+use Swoole\Server;
+
+
+/**
+ * Class OnServerTask
+ * @package Server\Task
+ */
+class OnServerTask
+{
+
+
+	#[Inject(Logger::class)]
+	public Logger $logger;
+
+
+	/**
+	 * @param Server $server
+	 * @param int $task_id
+	 * @param int $src_worker_id
+	 * @param mixed $data
+	 * @throws ConfigException
+	 */
+	public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data)
+	{
+		try {
+			$data = $this->resolve($data);
+		} catch (\Throwable $exception) {
+			$data = jTraceEx($exception);
+
+			$this->logger->error('task', [$data]);
+		} finally {
+			$server->finish($data);
+		}
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param Server\Task $task
+	 * @throws ConfigException
+	 */
+	public function onCoroutineTask(Server $server, Server\Task $task)
+	{
+		try {
+			$data = $this->resolve($task->data);
+		} catch (\Throwable $exception) {
+			$data = jTraceEx($exception);
+
+			$this->logger->error('task', [$data]);
+		} finally {
+			$server->finish($data);
+		}
+	}
+
+
+	/**
+	 * @param $data
+	 * @return null
+	 * @throws ReflectionException
+	 */
+	private function resolve($data)
+	{
+		[$class, $params] = json_encode($data, true);
+
+		$reflect = Kiri::getDi()->getReflect($class);
+
+		if (!$reflect->isInstantiable()) {
+			return null;
+		}
+		$class = $reflect->newInstanceArgs($params);
+		return $class->execute();
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param int $task_id
+	 * @param mixed $data
+	 */
+	public function onFinish(Server $server, int $task_id, mixed $data)
+	{
+		if (!($data instanceof OnTaskInterface)) {
+			return;
+		}
+		$data->finish($server, $task_id);
+	}
+
+
+}

Handler/OnServerWorker.php

@@ -0,0 +1,135 @@
+<?php
+
+namespace Server\Handler;
+
+use Annotation\Inject;
+use Exception;
+use Kiri\Abstracts\Config;
+use Kiri\Core\Help;
+use Kiri\Events\EventDispatch;
+use Kiri\Runtime;
+use Server\Events\OnAfterWorkerStart;
+use Server\Events\OnBeforeWorkerStart;
+use Server\Events\OnTaskerStart as OnTaskStart;
+use Server\Events\OnWorkerError;
+use Server\Events\OnWorkerExit;
+use Server\Events\OnWorkerStart;
+use Server\Events\OnWorkerStop;
+use Server\ServerManager;
+use Swoole\Server;
+use Swoole\Timer;
+
+
+/**
+ * Class OnServerWorker
+ * @package Server\Worker
+ */
+class OnServerWorker extends \Server\Abstracts\Server
+{
+
+
+	/**
+	 * @var EventDispatch
+	 */
+	#[Inject(EventDispatch::class)]
+	public EventDispatch $eventDispatch;
+
+
+	/**
+	 * @param Server $server
+	 * @param int $workerId
+	 * @throws Exception
+	 */
+	public function onWorkerStart(Server $server, int $workerId)
+	{
+		$this->eventDispatch->dispatch(new OnBeforeWorkerStart($workerId));
+		if ($workerId < $server->setting['worker_num']) {
+			$this->eventDispatch->dispatch(new OnWorkerStart($server, $workerId));
+			$this->setProcessName(sprintf('Worker[%d].%d', $server->worker_pid, $workerId));
+		} else {
+			$this->eventDispatch->dispatch(new OnTaskStart($server, $workerId));
+			$this->setProcessName(sprintf('Tasker[%d].%d', $server->worker_pid, $workerId));
+		}
+		$this->eventDispatch->dispatch(new OnAfterWorkerStart());
+	}
+
+
+	/**
+	 * @param OnBeforeWorkerStart $worker
+	 * @throws Exception
+	 */
+	public function setConfigure(OnBeforeWorkerStart $worker)
+	{
+		ServerManager::setEnv('worker', $worker->workerId);
+		$serialize = file_get_contents(storage(Runtime::CONFIG_NAME));
+		if (!empty($serialize)) {
+			Config::sets(unserialize($serialize));
+		}
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param int $workerId
+	 * @throws Exception
+	 */
+	public function onWorkerStop(Server $server, int $workerId)
+	{
+		$this->eventDispatch->dispatch(new OnWorkerStop($server, $workerId));
+
+		Timer::clearAll();
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param int $workerId
+	 * @throws Exception
+	 */
+	public function onWorkerExit(Server $server, int $workerId)
+	{
+		$this->eventDispatch->dispatch(new OnWorkerExit($server, $workerId));
+
+		ServerManager::setEnv('state', 'exit');
+	}
+
+
+	/**
+	 * @param Server $server
+	 * @param int $worker_id
+	 * @param int $worker_pid
+	 * @param int $exit_code
+	 * @param int $signal
+	 * @throws Exception
+	 */
+	public function onWorkerError(Server $server, int $worker_id, int $worker_pid, int $exit_code, int $signal)
+	{
+		$this->eventDispatch->dispatch(new OnWorkerError($server, $worker_id, $worker_pid, $exit_code, $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(), 9)
+		);
+
+		$this->logger->error($message);
+
+		$this->system_mail($message);
+	}
+
+
+	/**
+	 * @param $messageContent
+	 * @throws Exception
+	 */
+	protected function system_mail($messageContent)
+	{
+		try {
+			$email = Config::get('email');
+			if (!empty($email) && ($email['enable'] ?? false) == true) {
+				Help::sendEmail($email, 'Service Error', $messageContent);
+			}
+		} catch (\Throwable $e) {
+			error($e, 'email');
+		}
+	}
+
+}