改名

Tasker/AsyncTask.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Server;
+
+use Annotation\Inject;
+use Exception;
+use Kiri\Di\Container;
+use Psr\Container\ContainerInterface;
+use Server\Contract\OnTaskInterface;
+
+class AsyncTask
+{
+
+
+	/**
+	 * @var SwooleServerInterface
+	 */
+	#[Inject(SwooleServerInterface::class)]
+	public SwooleServerInterface $server;
+
+
+	/**
+	 * @var Container
+	 */
+	#[Inject(ContainerInterface::class)]
+	public ContainerInterface $container;
+
+
+	/**
+	 * @param OnTaskInterface|string $handler
+	 * @param array $params
+	 * @param int|null $workerId
+	 * @throws Exception
+	 */
+	public function execute(OnTaskInterface|string $handler, array $params = [], int $workerId = null)
+	{
+		if ($workerId === null || $workerId <= $this->server->setting['worker_num']) {
+			$workerId = random_int($this->server->setting['worker_num'] + 1,
+				$this->server->setting['worker_num'] + 1 + $this->server->setting['task_worker_num']);
+		}
+		if (is_string($handler)) {
+			$handler = $this->handle($handler, $params);
+		}
+		$this->server->task(serialize($handler), $workerId);
+	}
+
+
+	/**
+	 * @param $handler
+	 * @param $params
+	 * @return object
+	 * @throws \ReflectionException
+	 * @throws Exception
+	 */
+	private function handle($handler, $params): object
+	{
+		$implements = $this->container->getReflect($handler);
+		if (!in_array(OnTaskInterface::class, $implements->getInterfaceNames())) {
+			throw new Exception('Task must instance ' . OnTaskInterface::class);
+		}
+		return $implements->newInstanceArgs($params);
+	}
+
+
+}

Tasker/OnServerTask.php

@@ -0,0 +1,95 @@
+<?php
+
+
+namespace Server\Tasker;
+
+
+use Annotation\Inject;
+use Kiri\Abstracts\Logger;
+use Kiri\Exception\ConfigException;
+use ReflectionException;
+use Server\Contract\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
+	 */
+	private function resolve($data)
+	{
+		$execute = unserialize($data);
+		if ($execute instanceof OnTaskInterface) {
+			return $execute->execute();
+		}
+		return null;
+	}
+
+
+	/**
+	 * @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);
+	}
+
+
+}