as2252258/kiri-http-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

qqq

Browse Source
This commit is contained in:
向林
2023-08-17 16:56:51 +08:00
parent 8fe0453fe4
commit 8257e675ba
2 changed files with 106 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Task/Task.php
+86 -66
View File
@@ -25,78 +25,98 @@ class Task implements TaskInterface
/** /**
* @param Server $server * @return void
* @return void * @throws ReflectionException
*/ */
public function initTaskWorker(Server $server): void public function init(): void
{ {
if (!isset($server->setting[Constant::OPTION_TASK_WORKER_NUM])) { $this->server = Kiri::getDi()->get(ServerInterface::class);
return; }
}
if ($server->setting[Constant::OPTION_TASK_WORKER_NUM] < 1) {
return;
}
$server->on('finish', [$this, 'onFinish']);
$server->on('task', [$this, 'onTask']);
}
/** /**
* @param Server $server * @param Server $server
* @param int $task_id * @return void
* @param mixed $data */
* @return void public function initTaskWorker(Server $server): void
* @throws ContainerExceptionInterface {
* @throws NotFoundExceptionInterface if (!isset($server->setting[Constant::OPTION_TASK_WORKER_NUM])) {
* @throws ReflectionException return;
*/ }
public function onFinish(Server $server, int $task_id, mixed $data): void if ($server->setting[Constant::OPTION_TASK_WORKER_NUM] < 1) {
{ return;
event(new OnTaskFinish($task_id, $data)); }
} $server->on('finish', [$this, 'onFinish']);
$server->on('task', [$this, 'onTask']);
}
/** /**
* @param Server $server * @param Server $server
* @param int $task_id * @param int $task_id
* @param int $src_worker_id * @param mixed $data
* @param mixed $data * @return void
* @return mixed * @throws ContainerExceptionInterface
* @throws ReflectionException * @throws NotFoundExceptionInterface
*/ * @throws ReflectionException
public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): mixed */
{ public function onFinish(Server $server, int $task_id, mixed $data): void
$data = json_decode($data, true); {
if (is_null($data)) { event(new OnTaskFinish($task_id, $data));
return null; }
}
$data[0] = Kiri::getDi()->get($data[0]);
return call_user_func($data, $task_id, $src_worker_id);
}
/** /**
* @param array|string|object $handler * @param Server $server
* @param int|null $workerId * @param int $task_id
* @return void * @param int $src_worker_id
* @throws ReflectionException * @param mixed $data
*/ * @return mixed
public function dispatch(array|string|object $handler, ?int $workerId = null): void * @throws ReflectionException
{ */
if (is_null($workerId)) { public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): mixed
$workerId = rand(0, $this->server->setting[Constant::OPTION_TASK_WORKER_NUM] - 1); {
} $data = json_decode($data, true);
if (is_string($handler)) { if (is_null($data)) {
$this->server->task(serialize([di($handler), 'handle']), $workerId); return null;
} else if (is_array($handler)) { }
if (is_string($handler[0])) { $data[0] = Kiri::getDi()->get($data[0]);
$handler[0] = di($handler[0]); return call_user_func($data, $task_id, $src_worker_id);
} }
$this->server->task(serialize($handler), $workerId);
} else {
$this->server->task(serialize([$handler, 'handle']), $workerId); /**
} * @param mixed $data
} * @param float $timeout
* @param int $dstWorkerId
* @return mixed
*/
public function taskWait(mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed
{
return $this->server->taskwait($data, $timeout, $dstWorkerId);
}
/**
* @param array $tasks
* @param float $timeout
* @return false|array
*/
public function taskCo(array $tasks, float $timeout = 0.5): false|array
{
return $this->server->taskCo($tasks, $timeout);
}
/**
* @param array $tasks
* @param float $timeout
* @return false|array
*/
public function taskWaitMulti(array $tasks, float $timeout = 0.5): false|array
{
return $this->server->taskWaitMulti($tasks, $timeout);
}
} }
Task/TaskInterface.php
+20 -6
View File
@@ -6,12 +6,26 @@ interface TaskInterface
{ {
/** /**
* @param array $handler * @param array $tasks
* @param int|null $workerId * @param float $timeout
* @return void * @return false|array
*/ */
public function dispatch(array $handler, ?int $workerId = null): void; public function taskWaitMulti(array $tasks, float $timeout = 0.5): false|array;
/**
* @param array $tasks
* @param float $timeout
* @return false|array
*/
public function taskCo(array $tasks, float $timeout = 0.5): false|array;
/**
* @param mixed $data
* @param float $timeout
* @param int $dstWorkerId
* @return mixed
*/
public function taskWait(mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed;
} }