diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index 2973dd43..e26feb34 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -7,6 +7,7 @@ namespace HttpServer\Events; use Exception; use HttpServer\Abstracts\Callback; use Kafka\Struct; +use Snowflake\Crontab\Crontab; use Snowflake\Event; use Snowflake\Exception\ComponentException; use Swoole\Server; @@ -22,16 +23,15 @@ class OnPipeMessage extends Callback * @param Server $server * @param int $src_worker_id * @param $message - * @throws ComponentException * @throws Exception */ public function onHandler(Server $server, int $src_worker_id, $message) { try { - $swoole_unserialize = swoole_unserialize($message); - match ($swoole_unserialize['action'] ?? null) { - 'kafka' => $this->onKafkaWorker($swoole_unserialize), - 'crontab' => $this->onCrontabWorker($swoole_unserialize), + $swollen_universalize = swoole_unserialize($message); + match ($swollen_universalize['action'] ?? null) { + 'kafka' => $this->onKafkaWorker($swollen_universalize), + 'crontab' => $this->onCrontabWorker($swollen_universalize), default => $this->onMessageWorker($server, $src_worker_id, $message) }; } catch (\Throwable $exception) { @@ -42,26 +42,28 @@ class OnPipeMessage extends Callback } - /** - * @param array $message - * @return string - */ - private function onCrontabWorker(array $message) + /** + * @param array $message + * @return string + * @throws Exception + */ + private function onCrontabWorker(array $message): string { - /** @var \Snowflake\Crontab\Crontab $crontab */ + /** @var Crontab $crontab */ $crontab = $message['handler'] ?? null; $crontab->increment()->execute(); return 'success'; } - /** - * @param $server - * @param $src_worker_id - * @param $message - * @throws \Exception - */ - private function onMessageWorker($server, $src_worker_id, $message) + /** + * @param $server + * @param $src_worker_id + * @param $message + * @return string + * @throws Exception + */ + private function onMessageWorker($server, $src_worker_id, $message): string { fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); @@ -69,12 +71,11 @@ class OnPipeMessage extends Callback } - /** - * @param array $message - * @throws \ReflectionException - * @throws \Snowflake\Exception\NotFindClassException - */ - private function onKafkaWorker(array $message) + /** + * @param array $message + * @return string + */ + private function onKafkaWorker(array $message): string { [$topic, $rdMessage] = $message['body']; diff --git a/System/Crontab/Crontab.php b/System/Crontab/Crontab.php index a6dd74cc..0c92edf5 100644 --- a/System/Crontab/Crontab.php +++ b/System/Crontab/Crontab.php @@ -6,6 +6,7 @@ namespace Snowflake\Crontab; use Closure; use Exception; +use JetBrains\PhpStorm\Pure; use Snowflake\Abstracts\BaseObject; use Snowflake\Event; use Snowflake\Snowflake; @@ -64,9 +65,9 @@ class Crontab extends BaseObject /** * @return string */ - public function getName(): string + #[Pure] public function getName(): string { - return $this->name; + return md5($this->name); } /** @@ -203,7 +204,7 @@ class Crontab extends BaseObject { $redis = Snowflake::app()->getRedis(); - $redis->set('crontab:' . ($name = md5($this->getName())), swoole_serialize($this)); + $redis->set('crontab:' . ($name = $this->getName()), swoole_serialize($this)); $tickTime = time() + $this->getTickTime(); @@ -219,7 +220,7 @@ class Crontab extends BaseObject try { $redis = Snowflake::app()->getRedis(); - $name_md5 = md5($this->getName()); + $name_md5 = $this->getName(); $redis->hSet(self::WAIT_END, $name_md5, serialize($this)); $params = call_user_func($this->handler, $this->params, $this->name); diff --git a/System/Crontab/Producer.php b/System/Crontab/Producer.php index f7d5ea6b..1d1d7f7a 100644 --- a/System/Crontab/Producer.php +++ b/System/Crontab/Producer.php @@ -27,7 +27,7 @@ class Producer extends Component { $redis = Snowflake::app()->getRedis(); - $name = md5($crontab->getName()); + $name = $crontab->getName(); $redis->set('crontab:' . $name, swoole_serialize($crontab)); diff --git a/System/Crontab/Zookeeper.php b/System/Crontab/Zookeeper.php index 0d47ed5b..08c7c444 100644 --- a/System/Crontab/Zookeeper.php +++ b/System/Crontab/Zookeeper.php @@ -10,6 +10,7 @@ use Snowflake\Snowflake; use Swoole\Coroutine\WaitGroup; use Swoole\Coroutine\Channel; use Swoole\Timer; +use Snowflake\Cache\Redis; /** * Class Zookeeper @@ -43,7 +44,7 @@ class Zookeeper extends Process name($this->pid, 'Crontab zookeeper.'); } Timer::tick(1000, function () { - [$range, $redis] = $this->loadCrotabTask(); + [$range, $redis] = $this->loadCarobTask(); $server = Snowflake::app()->getSwoole(); $setting = $server->setting['worker_num']; @@ -55,12 +56,14 @@ class Zookeeper extends Process } - /** - * @param $server - * @param $redis - * @throws \Exception - */ - private function dispatch($server, $redis, $setting, $value) + /** + * @param $server + * @param Redis|\Redis $redis + * @param int $setting + * @param $value + * @throws Exception + */ + private function dispatch($server, Redis|\Redis $redis, int $setting, $value) { $server->sendMessage(swoole_serialize([ 'action' => 'crontab', 'handler' => swoole_unserialize($redis->get('crontab:' . $value)) @@ -71,9 +74,9 @@ class Zookeeper extends Process /** * @return array - * @throws \Exception + * @throws Exception */ - private function loadCrotabTask() + private function loadCarobTask(): array { $redis = Snowflake::app()->getRedis();