diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index 61de1d6e..98b4544b 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -20,82 +20,82 @@ use Swoole\Server; class OnPipeMessage extends Callback { - /** - * @param Server $server - * @param int $src_worker_id - * @param $swollen_universalize - * @throws Exception - */ - public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) - { - match ($swollen_universalize['action'] ?? null) { - 'kafka' => $this->onKafkaWorker($swollen_universalize), - 'crontab' => $this->onCrontabWorker($swollen_universalize), - default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) - }; - } + /** + * @param Server $server + * @param int $src_worker_id + * @param $swollen_universalize + * @throws Exception + */ + public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) + { + match ($swollen_universalize['action'] ?? null) { + 'kafka' => $this->onKafkaWorker($swollen_universalize), + 'crontab' => $this->onCrontabWorker($swollen_universalize), + default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) + }; + } - /** - * @param array $message - * @return string - * @throws Exception - */ - private function onCrontabWorker(array $message): string - { - if (empty($message['handler'] ?? null)) { - throw new Exception('unknown handler'); - } - /** @var Crontab $handler */ - $handler = $message['handler']; - defer(function () use ($handler) { - $return = $handler->isRecover(); - if ($return === 999) { - $name = $handler->getName(); + /** + * @param array $message + * @return string + * @throws Exception + */ + private function onCrontabWorker(array $message): string + { + if (empty($message['handler'] ?? null)) { + throw new Exception('unknown handler'); + } + /** @var Crontab $handler */ + $handler = swoole_unserialize($message['handler']); + defer(function () use ($handler) { + if ($handler->isRecover() !== 999) { + return; + } + $redis = Snowflake::app()->getRedis(); - $redis = Snowflake::app()->getRedis(); - if ($redis->exists('stop:crontab:' . $name)) { - $redis->del('crontab:' . $name); - $redis->del('stop:crontab:' . $name); - } else { - $redis->set('crontab:' . $name, swoole_serialize($handler)); - $tickTime = time() + $handler->getTickTime(); - $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); - } - } - }); - $handler->increment()->execute(); - return 'success'; - } + $name = $handler->getName(); + if (!$redis->exists('stop:crontab:' . $name)) { + $redis->set('crontab:' . $name, swoole_serialize($handler)); + $tickTime = time() + $handler->getTickTime(); + $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); + } else { + $redis->del('crontab:' . $name); + $redis->del('stop:crontab:' . $name); + } + }); + $handler->increment()->execute(); + return 'success'; + } - /** - * @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]); + /** + * @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]); - return 'success'; - } + return 'success'; + } - /** - * @param array $message - * @return string - */ - private function onKafkaWorker(array $message): string - { - [$topic, $rdMessage] = $message['body']; + /** + * @param array $message + * @return string + */ + private function onKafkaWorker(array $message): string + { + [$topic, $rdMessage] = $message['body']; - call_user_func($message['handler'], new Struct($topic, $rdMessage)); + call_user_func($message['handler'], new Struct($topic, $rdMessage)); - return 'success'; - } + return 'success'; + } } diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index e00e7ea7..17afb847 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -10,7 +10,8 @@ use RdKafka\ConsumerTopic; use RdKafka\Exception; use RdKafka\KafkaConsumer; use RdKafka\TopicConf; -use Snowflake\Core\Json; +use Snowflake\Abstracts\Config; +use Snowflake\Exception\ConfigException; use Snowflake\Snowflake; use Swoole\Coroutine\Channel; use Swoole\Coroutine\System; @@ -24,170 +25,198 @@ use Throwable; class Kafka extends \Snowflake\Process\Process { - protected Channel $channel; + protected Channel $channel; - private int $maxLength = 5000; + private int $maxLength = 5000; - /** - * @param Process $process - * @throws \Exception - */ - public function onHandler(Process $process): void - { - $content = System::readFile(storage('runtime.php')); - - $annotation = Snowflake::app()->getAnnotation(); - $annotation->setLoader(unserialize($content)); - $annotation->runtime(KAFKA_PATH); - - $this->waite(swoole_unserialize($process->read())); - } + /** + * @var array + */ + private array $kafkaConfig = []; - /** - * @param array $kafkaServer - * @throws \Exception - */ - private function waite(array $kafkaServer) - { - try { - name($this->pid, 'Kafka Consumer ' . $kafkaServer['topic']); + /** + * @return string + * @throws ConfigException + */ + public function getProcessName(): string + { + $this->kafkaConfig = swoole_unserialize($this->read()); - [$config, $topic, $conf] = $this->kafkaConfig($kafkaServer); - if (empty($config) && empty($topic) && empty($conf)) { - return; - } - $objRdKafka = new Consumer($config); - $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); + $name = Config::get('id', 'system') . '[' . $this->pid . ']'; - $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); - do { - $this->resolve($topic, $conf['interval'] ?? 1000); - } while (true); - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - } + return $name . '.' . 'Kafka Consumer ' . $this->kafkaConfig['topic']; + } - /** - * @param ConsumerTopic $topic - * @param $interval - * @throws \Exception - */ - private function resolve(ConsumerTopic $topic, $interval) - { - try { - $message = $topic->consume(0, $interval); - if (empty($message)) { - return; - } - if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) { - $this->handlerExecute($message->topic_name, $message); - } else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { - logger()->warning('No more messages; will wait for more'); - } else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) { - logger()->error('Kafka Timed out'); - } else { - logger()->error($message->errstr()); - } - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - } + /** + * @param Process $process + * @throws \Exception + */ + public function before(Process $process): void + { + $content = System::readFile(storage('runtime.php')); + + $annotation = Snowflake::app()->getAnnotation(); + $annotation->setLoader(unserialize($content)); + $annotation->runtime(KAFKA_PATH); + } - /** - * @param $topic - * @param $message - * @throws \Exception - */ - protected function handlerExecute($topic, $message) - { - go(function () use ($topic, $message) { - try { - $server = Snowflake::app()->getSwoole(); - - $setting = $server->setting['worker_num']; - - /** @var TaskContainer $container */ - $container = Snowflake::app()->get('kafka-container'); - $handler = $container->getConsumer($topic); - - if (empty($handler)) { - return; - } - - $message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]); - - $server->sendMessage($message, random_int(0, $setting - 1)); - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - }); - } + /** + * @param Process $process + * @throws \Exception + */ + public function onHandler(Process $process): void + { + $this->waite($this->kafkaConfig); + } - /** - * @param $kafka - * @return array - * @throws \Exception - */ - private function kafkaConfig($kafka): array - { - try { - $conf = new Conf(); - $conf->setRebalanceCb([$this, 'rebalanced_cb']); - $conf->set('group.id', $kafka['groupId']); - $conf->set('metadata.broker.list', $kafka['brokers']); - $conf->set('socket.timeout.ms', '30000'); + /** + * @param array $kafkaServer + * @throws \Exception + */ + private function waite(array $kafkaServer) + { + try { + name($this->pid, 'Kafka Consumer ' . $kafkaServer['topic']); - logger()->debug('kafka listen groupId ' . $kafka['groupId']); - logger()->debug('kafka listen brokers ' . $kafka['brokers']); + [$config, $topic, $conf] = $this->kafkaConfig($kafkaServer); + if (empty($config) && empty($topic) && empty($conf)) { + return; + } + $objRdKafka = new Consumer($config); + $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); - if (function_exists('pcntl_sigprocmask')) { - pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); - $conf->set('internal.termination.signal', (string)SIGIO); - } - - $topicConf = new TopicConf(); - $topicConf->set('auto.commit.enable', '1'); - $topicConf->set('auto.commit.interval.ms', '100'); - - //smallest：简单理解为从头开始消费， - //largest：简单理解为从最新的开始消费 - $topicConf->set('auto.offset.reset', 'smallest'); - $topicConf->set('offset.store.path', 'kafka_offset.log'); - $topicConf->set('offset.store.method', 'broker'); - - return [$conf, $topicConf, $kafka]; - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - - return [null, null, null]; - } - - } + $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); + do { + $this->resolve($topic, $conf['interval'] ?? 1000); + } while (true); + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + } - /** - * @param KafkaConsumer $kafka - * @param $err - * @param array|null $partitions - * @throws Exception - * @throws \Exception - */ - public function rebalanced_cb(KafkaConsumer $kafka, $err, array $partitions = null) - { - if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) { - $kafka->assign($partitions); - } else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) { - $kafka->assign(NULL); - } else { - throw new \Exception($err); - } - } + /** + * @param ConsumerTopic $topic + * @param $interval + * @throws \Exception + */ + private function resolve(ConsumerTopic $topic, $interval) + { + try { + $message = $topic->consume(0, $interval); + if (empty($message)) { + return; + } + if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) { + $this->handlerExecute($message->topic_name, $message); + } else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { + logger()->warning('No more messages; will wait for more'); + } else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) { + logger()->error('Kafka Timed out'); + } else { + logger()->error($message->errstr()); + } + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + } + + + /** + * @param $topic + * @param $message + * @throws \Exception + */ + protected function handlerExecute($topic, $message) + { + go(function () use ($topic, $message) { + try { + $server = Snowflake::app()->getSwoole(); + + $setting = $server->setting['worker_num']; + + /** @var TaskContainer $container */ + $container = Snowflake::app()->get('kafka-container'); + $handler = $container->getConsumer($topic); + + if (empty($handler)) { + return; + } + + $message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]); + + $server->sendMessage($message, random_int(0, $setting - 1)); + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + }); + } + + + /** + * @param $kafka + * @return array + * @throws \Exception + */ + private function kafkaConfig($kafka): array + { + try { + $conf = new Conf(); + $conf->setRebalanceCb([$this, 'rebalanced_cb']); + $conf->set('group.id', $kafka['groupId']); + $conf->set('metadata.broker.list', $kafka['brokers']); + $conf->set('socket.timeout.ms', '30000'); + + logger()->debug('kafka listen groupId ' . $kafka['groupId']); + logger()->debug('kafka listen brokers ' . $kafka['brokers']); + + if (function_exists('pcntl_sigprocmask')) { + pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); + $conf->set('internal.termination.signal', (string)SIGIO); + } + + $topicConf = new TopicConf(); + $topicConf->set('auto.commit.enable', '1'); + $topicConf->set('auto.commit.interval.ms', '100'); + + //smallest：简单理解为从头开始消费， + //largest：简单理解为从最新的开始消费 + $topicConf->set('auto.offset.reset', 'smallest'); + $topicConf->set('offset.store.path', 'kafka_offset.log'); + $topicConf->set('offset.store.method', 'broker'); + + return [$conf, $topicConf, $kafka]; + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + + return [null, null, null]; + } + + } + + + /** + * @param KafkaConsumer $kafka + * @param $err + * @param array|null $partitions + * @throws Exception + * @throws \Exception + */ + public function rebalanced_cb(KafkaConsumer $kafka, $err, array $partitions = null) + { + if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) { + $kafka->assign($partitions); + } else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) { + $kafka->assign(NULL); + } else { + throw new \Exception($err); + } + } } diff --git a/System/Crontab/Zookeeper.php b/System/Crontab/Zookeeper.php index 4546c699..b7bcd0c8 100644 --- a/System/Crontab/Zookeeper.php +++ b/System/Crontab/Zookeeper.php @@ -8,6 +8,7 @@ use Exception; use HttpServer\Server; use Snowflake\Abstracts\Config; use Snowflake\Cache\Redis; +use Snowflake\Exception\ConfigException; use Snowflake\Process\Process; use Snowflake\Snowflake; use Swoole\Coroutine; @@ -27,9 +28,10 @@ class Zookeeper extends Process private mixed $server; - /** - * @param \Swoole\Process $process - */ + /** + * @return string + * @throws ConfigException + */ public function getProcessName(): string { $name = Config::get('id', 'system') . '[' . $this->pid . ']'; @@ -85,7 +87,7 @@ class Zookeeper extends Process if (empty($handler = $redis->get('crontab:' . $value))) { return; } - $params['handler'] = unserialize($handler); + $params['handler'] = $handler; $this->server->sendMessage($params, $this->getWorker()); } catch (Throwable $exception) { @@ -104,10 +106,10 @@ class Zookeeper extends Process } - /** - * @return array - * @throws Exception - */ + /** + * @param Redis|\Redis $redis + * @return array + */ private function loadCarobTask(Redis|\Redis $redis): array { $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time()); diff --git a/System/Error/LoggerProcess.php b/System/Error/LoggerProcess.php index 576f4e86..445ba4a7 100644 --- a/System/Error/LoggerProcess.php +++ b/System/Error/LoggerProcess.php @@ -21,6 +21,19 @@ class LoggerProcess extends Process { + public function before(\Swoole\Process $process): void + { + // TODO: Implement before() method. + } + + + + public function getProcessName(): string + { + // TODO: Implement getProcessName() method. + } + + /** * @param \Swoole\Process $process * @throws ComponentException diff --git a/System/Process/Biomonitoring.php b/System/Process/Biomonitoring.php index 48d10e4b..c8044bfb 100644 --- a/System/Process/Biomonitoring.php +++ b/System/Process/Biomonitoring.php @@ -5,6 +5,7 @@ namespace Snowflake\Process; use Exception; +use JetBrains\PhpStorm\Pure; use Snowflake\Snowflake; use Swoole\Timer; @@ -16,6 +17,25 @@ class Biomonitoring extends Process { + /** + * @param \Swoole\Process $process + */ + public function before(\Swoole\Process $process): void + { + // TODO: Implement before() method. + } + + + /** + * @return string + */ + #[Pure] public function getProcessName(): string + { + // TODO: Implement getProcessName() method. + return get_called_class(); + } + + /** * @param \Swoole\Process $process * @throws Exception diff --git a/System/Process/SProcess.php b/System/Process/SProcess.php index afa998bf..da003d5d 100644 --- a/System/Process/SProcess.php +++ b/System/Process/SProcess.php @@ -8,10 +8,18 @@ interface SProcess { -// public function getProcessName(): string; -// -// -// public function before(\Swoole\Process $process): void; + /** + * @return string + * + * return process name + */ + public function getProcessName(): string; + + + /** + * @param \Swoole\Process $process + */ + public function before(\Swoole\Process $process): void; /** * @param \Swoole\Process $process diff --git a/System/Process/ServerInotify.php b/System/Process/ServerInotify.php index 38387366..7a8f93a6 100644 --- a/System/Process/ServerInotify.php +++ b/System/Process/ServerInotify.php @@ -12,10 +12,8 @@ namespace Snowflake\Process; use Exception; use Snowflake\Abstracts\Config; -use Snowflake\Exception\ComponentException; use Snowflake\Exception\ConfigException; use Snowflake\Snowflake; -use Swoole\Coroutine; use Swoole\Event; use Swoole\Timer; @@ -25,290 +23,310 @@ use Swoole\Timer; */ class ServerInotify extends Process { - private mixed $inotify; - private bool $isReloading = false; - private bool $isReloadingOut = false; - private array $watchFiles = []; - private ?array $dirs = []; - private int $events; + private mixed $inotify; + private bool $isReloading = false; + private bool $isReloadingOut = false; + private array $watchFiles = []; + private ?array $dirs = []; + private int $events; - private int $int = -1; + private int $int = -1; - /** - * @param \Swoole\Process $process - * @throws Exception - */ - public function onHandler(\Swoole\Process $process): void - { - set_error_handler([$this, 'onErrorHandler']); - $this->dirs = Config::get('inotify', [APP_PATH]); - if (extension_loaded('inotify')) { - $this->inotify = inotify_init(); - $this->events = IN_MODIFY | IN_DELETE | IN_CREATE | IN_MOVE; - - foreach ($this->dirs as $dir) { - if (!is_dir($dir)) continue; - $this->watch($dir); - } - Event::add($this->inotify, [$this, 'check']); - Event::wait(); - } else { - $this->loadDirs(); - $this->tick(); - } - } + /** + * @param \Swoole\Process $process + * @throws ConfigException + */ + public function before(\Swoole\Process $process): void + { + // TODO: Implement before() method. + set_error_handler([$this, 'onErrorHandler']); + $this->dirs = Config::get('inotify', [APP_PATH]); + if (extension_loaded('inotify')) { + $this->inotify = inotify_init(); + $this->events = IN_MODIFY | IN_DELETE | IN_CREATE | IN_MOVE; + } + } - /** - * @param bool $isReload - * @throws Exception - */ - private function loadDirs($isReload = false) - { - foreach ($this->dirs as $value) { - if (is_bool($path = realpath($value))) { - continue; - } - - if (!is_dir($path)) continue; - - $this->loadByDir($path, $isReload); - } - } + /** + * @return string + */ + public function getProcessName(): string + { + return 'file change process'; + } - private array $md5Map = []; + /** + * @param \Swoole\Process $process + * @throws Exception + */ + public function onHandler(\Swoole\Process $process): void + { + if (extension_loaded('inotify')) { + foreach ($this->dirs as $dir) { + if (!is_dir($dir)) continue; + $this->watch($dir); + } + Event::add($this->inotify, [$this, 'check']); + Event::wait(); + } else { + $this->loadDirs(); + $this->tick(); + } + } - /** - * @throws Exception - */ - public function tick() - { - if ($this->isReloading) { - return; - } + /** + * @param bool $isReload + * @throws Exception + */ + private function loadDirs($isReload = false) + { + foreach ($this->dirs as $value) { + if (is_bool($path = realpath($value))) { + continue; + } - $this->loadDirs(true); + if (!is_dir($path)) continue; - Timer::after(2000, [$this, 'tick']); - } + $this->loadByDir($path, $isReload); + } + } - /** - * @param $path - * @param bool $isReload - * @return void - * @throws Exception - */ - private function loadByDir($path, $isReload = false): void - { - if (!is_string($path)) { - return; - } - $path = rtrim($path, '/'); - foreach (glob(realpath($path) . '/*') as $value) { - if (is_dir($value)) { - $this->loadByDir($value, $isReload); - } - if (is_file($value)) { - if ($this->checkFile($value, $isReload)) { - $this->timerReload(); - break; - } - } - } - } + private array $md5Map = []; - /** - * @param $value - * @param $isReload - * @return bool - */ - private function checkFile($value, $isReload): bool - { - $md5 = md5($value); - $mTime = filectime($value); - if (!isset($this->md5Map[$md5])) { - if ($isReload) { - return true; - } - $this->md5Map[$md5] = $mTime; - } else { - if ($this->md5Map[$md5] != $mTime) { - if ($isReload) { - return true; - } - $this->md5Map[$md5] = $mTime; - } - } - return false; - } + /** + * @throws Exception + */ + public function tick() + { + if ($this->isReloading) { + return; + } + + $this->loadDirs(true); + + Timer::after(2000, [$this, 'tick']); + } - /** - * 开始监听 - */ - public function check() - { - if (!($events = inotify_read($this->inotify))) { - return; - } - if ($this->isReloading) { - if (!$this->isReloadingOut) { - $this->isReloadingOut = true; - } - return; - } - - $eventList = [IN_CREATE, IN_DELETE, IN_MODIFY, IN_MOVED_TO, IN_MOVED_FROM]; - foreach ($events as $ev) { - if (empty($ev['name'])) { - continue; - } - if ($ev['mask'] == IN_IGNORED) { - continue; - } - if (!in_array($ev['mask'], $eventList)) { - continue; - } - $fileType = strstr($ev['name'], '.'); - //非重启类型 - if ($fileType !== '.php') { - continue; - } - if ($this->int !== -1) { - return; - } - $this->int = @swoole_timer_after(2000, [$this, 'reload']); - - $this->isReloading = true; - } - } - - /** - * @throws Exception - */ - public function reload() - { - $this->isReloading = true; - $this->trigger_reload(); - - $this->clearWatch(); - foreach ($this->dirs as $root) { - $this->watch($root); - } - $this->int = -1; - $this->isReloading = FALSE; - $this->isReloadingOut = FALSE; - $this->md5Map = []; - } - - /** - * @throws Exception - */ - public function timerReload() - { - $this->isReloading = true; - $this->trigger_reload(); - - $this->int = -1; - - $this->loadDirs(); - - $this->isReloading = FALSE; - $this->isReloadingOut = FALSE; - - $this->tick(); - } + /** + * @param $path + * @param bool $isReload + * @return void + * @throws Exception + */ + private function loadByDir($path, $isReload = false): void + { + if (!is_string($path)) { + return; + } + $path = rtrim($path, '/'); + foreach (glob(realpath($path) . '/*') as $value) { + if (is_dir($value)) { + $this->loadByDir($value, $isReload); + } + if (is_file($value)) { + if ($this->checkFile($value, $isReload)) { + $this->timerReload(); + break; + } + } + } + } - /** - * 重启 - * @throws Exception - */ - public function trigger_reload() - { - exec(PHP_BINARY . ' snowflake runtime:builder'); - - Snowflake::reload(); - } - - /** - * @throws Exception - */ - public function clearWatch() - { - foreach ($this->watchFiles as $wd) { - try { - inotify_rm_watch($this->inotify, $wd); - } catch (\Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - } - $this->watchFiles = []; - } + /** + * @param $value + * @param $isReload + * @return bool + */ + private function checkFile($value, $isReload): bool + { + $md5 = md5($value); + $mTime = filectime($value); + if (!isset($this->md5Map[$md5])) { + if ($isReload) { + return true; + } + $this->md5Map[$md5] = $mTime; + } else { + if ($this->md5Map[$md5] != $mTime) { + if ($isReload) { + return true; + } + $this->md5Map[$md5] = $mTime; + } + } + return false; + } - /** - * @param $code - * @param $message - * @param $file - * @param $line - * @throws Exception - */ - protected function onErrorHandler($code, $message, $file, $line) - { - if (str_contains($message, 'The file descriptor is not an inotify instance')) { - return; - } - logger()->debug('Error:' . $message); - logger()->debug($file . ':' . $line); - } + /** + * 开始监听 + */ + public function check() + { + if (!($events = inotify_read($this->inotify))) { + return; + } + if ($this->isReloading) { + if (!$this->isReloadingOut) { + $this->isReloadingOut = true; + } + return; + } + + $eventList = [IN_CREATE, IN_DELETE, IN_MODIFY, IN_MOVED_TO, IN_MOVED_FROM]; + foreach ($events as $ev) { + if (empty($ev['name'])) { + continue; + } + if ($ev['mask'] == IN_IGNORED) { + continue; + } + if (!in_array($ev['mask'], $eventList)) { + continue; + } + $fileType = strstr($ev['name'], '.'); + //非重启类型 + if ($fileType !== '.php') { + continue; + } + if ($this->int !== -1) { + return; + } + $this->int = @swoole_timer_after(2000, [$this, 'reload']); + + $this->isReloading = true; + } + } + + /** + * @throws Exception + */ + public function reload() + { + $this->isReloading = true; + $this->trigger_reload(); + + $this->clearWatch(); + foreach ($this->dirs as $root) { + $this->watch($root); + } + $this->int = -1; + $this->isReloading = FALSE; + $this->isReloadingOut = FALSE; + $this->md5Map = []; + } + + /** + * @throws Exception + */ + public function timerReload() + { + $this->isReloading = true; + $this->trigger_reload(); + + $this->int = -1; + + $this->loadDirs(); + + $this->isReloading = FALSE; + $this->isReloadingOut = FALSE; + + $this->tick(); + } - /** - * @param $dir - * @return bool - * @throws Exception - */ - public function watch($dir): bool - { - //目录不存在 - if (!is_dir($dir)) { - return logger()->addError("[$dir] is not a directory."); - } - //避免重复监听 - if (isset($this->watchFiles[$dir])) { - return FALSE; - } + /** + * 重启 + * @throws Exception + */ + public function trigger_reload() + { + exec(PHP_BINARY . ' snowflake runtime:builder'); - if (in_array($dir, [APP_PATH . 'config', APP_PATH . 'commands', APP_PATH . '.git', APP_PATH . '.gitee'])) { - return FALSE; - } + Snowflake::reload(); + } - $wd = @inotify_add_watch($this->inotify, $dir, $this->events); - $this->watchFiles[$dir] = $wd; + /** + * @throws Exception + */ + public function clearWatch() + { + foreach ($this->watchFiles as $wd) { + try { + inotify_rm_watch($this->inotify, $wd); + } catch (\Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + } + $this->watchFiles = []; + } - $files = scandir($dir); - foreach ($files as $f) { - if ($f == '.' or $f == '..' or $f == 'runtime' or preg_match('/\.txt/', $f) or preg_match('/\.sql/', $f) or preg_match('/\.log/', $f)) { - continue; - } - $path = $dir . '/' . $f; - //递归目录 - if (is_dir($path)) { - $this->watch($path); - } - //检测文件类型 - if (strstr($f, '.') == '.php') { - $wd = @inotify_add_watch($this->inotify, $path, $this->events); - $this->watchFiles[$path] = $wd; - } - } - return TRUE; - } + /** + * @param $code + * @param $message + * @param $file + * @param $line + * @throws Exception + */ + protected function onErrorHandler($code, $message, $file, $line) + { + if (str_contains($message, 'The file descriptor is not an inotify instance')) { + return; + } + logger()->debug('Error:' . $message); + logger()->debug($file . ':' . $line); + } + + + /** + * @param $dir + * @return bool + * @throws Exception + */ + public function watch($dir): bool + { + //目录不存在 + if (!is_dir($dir)) { + return logger()->addError("[$dir] is not a directory."); + } + //避免重复监听 + if (isset($this->watchFiles[$dir])) { + return FALSE; + } + + if (in_array($dir, [APP_PATH . 'config', APP_PATH . 'commands', APP_PATH . '.git', APP_PATH . '.gitee'])) { + return FALSE; + } + + $wd = @inotify_add_watch($this->inotify, $dir, $this->events); + $this->watchFiles[$dir] = $wd; + + $files = scandir($dir); + foreach ($files as $f) { + if ($f == '.' or $f == '..' or $f == 'runtime' or preg_match('/\.txt/', $f) or preg_match('/\.sql/', $f) or preg_match('/\.log/', $f)) { + continue; + } + $path = $dir . '/' . $f; + //递归目录 + if (is_dir($path)) { + $this->watch($path); + } + + //检测文件类型 + if (strstr($f, '.') == '.php') { + $wd = @inotify_add_watch($this->inotify, $path, $this->events); + $this->watchFiles[$path] = $wd; + } + } + return TRUE; + } }