From 0aa3bc64487e95a8f7c5702ed48bb0c0a00cf00e Mon Sep 17 00:00:00 2001 From: millken Date: Thu, 3 Sep 2020 09:04:49 +0800 Subject: [PATCH] change swoole to 4.4.20 --- src/Application/Swoole.php | 6 +++--- src/Database/Connection.php | 4 ++-- src/Swoole/CronManager.php | 25 ++++++++++++------------- src/Swoole/Tasks/Server.php | 24 +++++++++++------------- src/Swoole/Tasks/Task.php | 26 ++++++++++++-------------- 5 files changed, 40 insertions(+), 45 deletions(-) diff --git a/src/Application/Swoole.php b/src/Application/Swoole.php index de65e2e..4927747 100644 --- a/src/Application/Swoole.php +++ b/src/Application/Swoole.php @@ -98,10 +98,10 @@ public function onTask(SwooleHttpServer $server, int $task_id, int $source, stri // 'data' => $data, // ]); - $task = SwooleSerialize::unpack($data); + $task = unserialize($data); $unit = Application::getContainer()->get($task->getClass()); - $result = $unit->run($task->getPayload()); - $server->finish(SwooleSerialize::pack($result, 1)); + $result = call_user_func([$unit, $task->getMethod()], $task->getParameter()); + $server->finish(serialize($result)); return $result; } diff --git a/src/Database/Connection.php b/src/Database/Connection.php index 5e741dd..f8cf810 100644 --- a/src/Database/Connection.php +++ b/src/Database/Connection.php @@ -430,7 +430,7 @@ public function reconnect() return $this->connect(); } - public function listen($callback) + public function listen(callable $callback): void { $this->listen = $callback; } @@ -440,7 +440,7 @@ protected function connection() return $this->pdo instanceof PDO ? $this->pdo : $this->connect(); } - public function action($actions) + public function action(callable $actions) { if (is_callable($actions)) { $this->connection()->beginTransaction(); diff --git a/src/Swoole/CronManager.php b/src/Swoole/CronManager.php index cb03b14..228ac66 100644 --- a/src/Swoole/CronManager.php +++ b/src/Swoole/CronManager.php @@ -1,14 +1,14 @@ isSubclassOf(CronWorker::class)) { - throw new Exception('cron worker mustbe extends \Ypf\Controller\\'.CronWorker::class); + throw new Exception('cron worker mustbe extends \Ypf\Controller\\' . CronWorker::class); } if (!isset($worker[1])) { go(function () use ($classReflection) { $classReflection->newInstance()->run(); }); } else { - $this->queue[] = [$classReflection->newInstance(), $worker[1]]; + if(is_int($worker[1])) { + \swoole_timer_tick(1000 * $worker[1], [$classReflection->newInstance(), 'run']); + }else{ + $this->queue[] = [$classReflection->newInstance(), $worker[1]]; + } } } \swoole_timer_tick(1000, [$this, 'tick']); @@ -46,14 +50,9 @@ public function tick() { $queue = $this->queue; foreach ($queue as $key => $val) { - $crontab = CronExpression::isValidExpression($val[1]); - if (!$crontab) { - $timeSecond = intval($val[1]); - } else { - $cron = CronExpression::factory($val[1]); - $nextRunTime = $cron->getNextRunDate()->getTimestamp(); - $timeSecond = intval($nextRunTime - time()); - } + $cron = CronExpression::factory($val[1]); + $nextRunTime = $cron->getNextRunDate()->getTimestamp(); + $timeSecond = intval($nextRunTime - time()); if ($timeSecond < 1) { continue; } @@ -61,7 +60,7 @@ public function tick() \swoole_timer_after(1000 * $timeSecond, function () use ($key, $val) { $this->queue[$key] = $val; unset($this->job[$key]); - go(function () use ($val) {$val[0]->run(); }); + go(function () use ($val) {$val[0]->run();}); }); unset($this->queue[$key]); $this->job[$key] = $val; diff --git a/src/Swoole/Tasks/Server.php b/src/Swoole/Tasks/Server.php index c078df7..11b64be 100644 --- a/src/Swoole/Tasks/Server.php +++ b/src/Swoole/Tasks/Server.php @@ -1,30 +1,28 @@ push($task); + */ public function push(Task $task) { - $payload = Serialize::pack($task, 1); - YAS::getServer()->task( - $payload, - -1, - function (SwooleServer $server, $source, $data) use ($task) { - call_user_func($task->getCallback(), Serialize::unpack($data)); - } - ); + $payload = serialize($task); + YAS::getServer()->task($payload, -1); } public function await(Task $task, float $timeout = 1) { - $payload = Serialize::pack($task, 1); + $payload = serialize($task); return YAS::getServer()->taskwait($payload, (float) $timeout); } @@ -35,7 +33,7 @@ public function parallel(array $tasks, float $timeout = 10): array $results = []; foreach ($tasks as $idx => $task) { /* @var Task $task */ - $normalized[] = Serialize::pack($task, 1); + $normalized[] = serialize($task); $results[$idx] = false; } $result = YAS::getServer()->taskWaitMulti($normalized, (float) $timeout); diff --git a/src/Swoole/Tasks/Task.php b/src/Swoole/Tasks/Task.php index 043ab90..da78352 100644 --- a/src/Swoole/Tasks/Task.php +++ b/src/Swoole/Tasks/Task.php @@ -7,13 +7,13 @@ class Task { private $class; - private $payload = null; - private $callback; + private $parameter = null; + private $method; - public function __construct(string $class, callable $callback = null) + public function __construct(string $class, string $method = null) { $this->class = $class; - $this->callback = $callback; + $this->method = $method; } public function getClass(): string @@ -21,25 +21,23 @@ public function getClass(): string return $this->class; } - public function withPayload($payload): void + public function withParameter($parameter): void { - $this->payload = $payload; + $this->parameter = $parameter; } - public function getPayload() + public function getParameter() { - return $this->payload; + return $this->parameter; } - public function withCallback(callable $callback): void + public function withMethod(string $method): void { - $this->callback = $callback; + $this->method = $method; } - public function getCallback(): callable + public function getMethod(): string { - return $this->callback ?? function () { - // Nothing to do - }; + return $this->method; } }