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

改名

Browse Source
This commit is contained in:
Mr·x
2021-04-26 17:34:33 +08:00
parent 76f9688ea6
commit 52357f8283
7 changed files with 577 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnPipeMessage.php
+66 -66
View File
@@ -20,82 +20,82 @@ use Swoole\Server;
class OnPipeMessage extends Callback class OnPipeMessage extends Callback
{ {
/** /**
* @param Server $server * @param Server $server
* @param int $src_worker_id * @param int $src_worker_id
* @param $swollen_universalize * @param $swollen_universalize
* @throws Exception * @throws Exception
*/ */
public function onHandler(Server $server, int $src_worker_id, $swollen_universalize) public function onHandler(Server $server, int $src_worker_id, $swollen_universalize)
{ {
match ($swollen_universalize['action'] ?? null) { match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize), 'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize), 'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize) default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize)
}; };
} }
/** /**
* @param array $message * @param array $message
* @return string * @return string
* @throws Exception * @throws Exception
*/ */
private function onCrontabWorker(array $message): string private function onCrontabWorker(array $message): string
{ {
if (empty($message['handler'] ?? null)) { if (empty($message['handler'] ?? null)) {
throw new Exception('unknown handler'); throw new Exception('unknown handler');
} }
/** @var Crontab $handler */ /** @var Crontab $handler */
$handler = $message['handler']; $handler = swoole_unserialize($message['handler']);
defer(function () use ($handler) { defer(function () use ($handler) {
$return = $handler->isRecover(); if ($handler->isRecover() !== 999) {
if ($return === 999) { return;
$name = $handler->getName(); }
$redis = Snowflake::app()->getRedis();
$redis = Snowflake::app()->getRedis(); $name = $handler->getName();
if ($redis->exists('stop:crontab:' . $name)) { if (!$redis->exists('stop:crontab:' . $name)) {
$redis->del('crontab:' . $name); $redis->set('crontab:' . $name, swoole_serialize($handler));
$redis->del('stop:crontab:' . $name); $tickTime = time() + $handler->getTickTime();
} else { $redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name);
$redis->set('crontab:' . $name, swoole_serialize($handler)); } else {
$tickTime = time() + $handler->getTickTime(); $redis->del('crontab:' . $name);
$redis->zAdd(Producer::CRONTAB_KEY, $tickTime, $name); $redis->del('stop:crontab:' . $name);
} }
} });
}); $handler->increment()->execute();
$handler->increment()->execute(); return 'success';
return 'success'; }
}
/** /**
* @param $server * @param $server
* @param $src_worker_id * @param $src_worker_id
* @param $message * @param $message
* @return string * @return string
* @throws Exception * @throws Exception
*/ */
private function onMessageWorker($server, $src_worker_id, $message): string private function onMessageWorker($server, $src_worker_id, $message): string
{ {
fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]);
return 'success'; return 'success';
} }
/** /**
* @param array $message * @param array $message
* @return string * @return string
*/ */
private function onKafkaWorker(array $message): string private function onKafkaWorker(array $message): string
{ {
[$topic, $rdMessage] = $message['body']; [$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';
} }
} }
Kafka/Kafka.php
+177 -148
View File
@@ -10,7 +10,8 @@ use RdKafka\ConsumerTopic;
use RdKafka\Exception; use RdKafka\Exception;
use RdKafka\KafkaConsumer; use RdKafka\KafkaConsumer;
use RdKafka\TopicConf; use RdKafka\TopicConf;
use Snowflake\Core\Json; use Snowflake\Abstracts\Config;
use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake; use Snowflake\Snowflake;
use Swoole\Coroutine\Channel; use Swoole\Coroutine\Channel;
use Swoole\Coroutine\System; use Swoole\Coroutine\System;
@@ -24,170 +25,198 @@ use Throwable;
class Kafka extends \Snowflake\Process\Process class Kafka extends \Snowflake\Process\Process
{ {
protected Channel $channel; protected Channel $channel;
private int $maxLength = 5000; private int $maxLength = 5000;
/** /**
* @param Process $process * @var array
* @throws \Exception */
*/ private array $kafkaConfig = [];
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()));
}
/** /**
* @param array $kafkaServer * @return string
* @throws \Exception * @throws ConfigException
*/ */
private function waite(array $kafkaServer) public function getProcessName(): string
{ {
try { $this->kafkaConfig = swoole_unserialize($this->read());
name($this->pid, 'Kafka Consumer ' . $kafkaServer['topic']);
[$config, $topic, $conf] = $this->kafkaConfig($kafkaServer); $name = Config::get('id', 'system') . '[' . $this->pid . ']';
if (empty($config) && empty($topic) && empty($conf)) {
return;
}
$objRdKafka = new Consumer($config);
$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); return $name . '.' . 'Kafka Consumer ' . $this->kafkaConfig['topic'];
do { }
$this->resolve($topic, $conf['interval'] ?? 1000);
} while (true);
} catch (Throwable $exception) {
logger()->addError($exception, 'throwable');
}
}
/** /**
* @param ConsumerTopic $topic * @param Process $process
* @param $interval * @throws \Exception
* @throws \Exception */
*/ public function before(Process $process): void
private function resolve(ConsumerTopic $topic, $interval) {
{ $content = System::readFile(storage('runtime.php'));
try {
$message = $topic->consume(0, $interval); $annotation = Snowflake::app()->getAnnotation();
if (empty($message)) { $annotation->setLoader(unserialize($content));
return; $annotation->runtime(KAFKA_PATH);
} }
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 Process $process
* @param $message * @throws \Exception
* @throws \Exception */
*/ public function onHandler(Process $process): void
protected function handlerExecute($topic, $message) {
{ $this->waite($this->kafkaConfig);
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 * @param array $kafkaServer
* @return array * @throws \Exception
* @throws \Exception */
*/ private function waite(array $kafkaServer)
private function kafkaConfig($kafka): array {
{ try {
try { name($this->pid, 'Kafka Consumer ' . $kafkaServer['topic']);
$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']); [$config, $topic, $conf] = $this->kafkaConfig($kafkaServer);
logger()->debug('kafka listen brokers ' . $kafka['brokers']); if (empty($config) && empty($topic) && empty($conf)) {
return;
}
$objRdKafka = new Consumer($config);
$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
if (function_exists('pcntl_sigprocmask')) { $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); do {
$conf->set('internal.termination.signal', (string)SIGIO); $this->resolve($topic, $conf['interval'] ?? 1000);
} } while (true);
} catch (Throwable $exception) {
$topicConf = new TopicConf(); logger()->addError($exception, 'throwable');
$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 ConsumerTopic $topic
* @param $err * @param $interval
* @param array|null $partitions * @throws \Exception
* @throws Exception */
* @throws \Exception private function resolve(ConsumerTopic $topic, $interval)
*/ {
public function rebalanced_cb(KafkaConsumer $kafka, $err, array $partitions = null) try {
{ $message = $topic->consume(0, $interval);
if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) { if (empty($message)) {
$kafka->assign($partitions); return;
} else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) { }
$kafka->assign(NULL); if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
} else { $this->handlerExecute($message->topic_name, $message);
throw new \Exception($err); } 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);
}
}
} }
System/Crontab/Zookeeper.php
+10 -8
View File
@@ -8,6 +8,7 @@ use Exception;
use HttpServer\Server; use HttpServer\Server;
use Snowflake\Abstracts\Config; use Snowflake\Abstracts\Config;
use Snowflake\Cache\Redis; use Snowflake\Cache\Redis;
use Snowflake\Exception\ConfigException;
use Snowflake\Process\Process; use Snowflake\Process\Process;
use Snowflake\Snowflake; use Snowflake\Snowflake;
use Swoole\Coroutine; use Swoole\Coroutine;
@@ -27,9 +28,10 @@ class Zookeeper extends Process
private mixed $server; private mixed $server;
/** /**
* @param \Swoole\Process $process * @return string
*/ * @throws ConfigException
*/
public function getProcessName(): string public function getProcessName(): string
{ {
$name = Config::get('id', 'system') . '[' . $this->pid . ']'; $name = Config::get('id', 'system') . '[' . $this->pid . ']';
@@ -85,7 +87,7 @@ class Zookeeper extends Process
if (empty($handler = $redis->get('crontab:' . $value))) { if (empty($handler = $redis->get('crontab:' . $value))) {
return; return;
} }
$params['handler'] = unserialize($handler); $params['handler'] = $handler;
$this->server->sendMessage($params, $this->getWorker()); $this->server->sendMessage($params, $this->getWorker());
} catch (Throwable $exception) { } catch (Throwable $exception) {
@@ -104,10 +106,10 @@ class Zookeeper extends Process
} }
/** /**
* @return array * @param Redis|\Redis $redis
* @throws Exception * @return array
*/ */
private function loadCarobTask(Redis|\Redis $redis): array private function loadCarobTask(Redis|\Redis $redis): array
{ {
$range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time()); $range = $redis->zRangeByScore(Producer::CRONTAB_KEY, '0', (string)time());
System/Error/LoggerProcess.php
+13
View File
@@ -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 * @param \Swoole\Process $process
* @throws ComponentException * @throws ComponentException
System/Process/Biomonitoring.php
+20
View File
@@ -5,6 +5,7 @@ namespace Snowflake\Process;
use Exception; use Exception;
use JetBrains\PhpStorm\Pure;
use Snowflake\Snowflake; use Snowflake\Snowflake;
use Swoole\Timer; 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 * @param \Swoole\Process $process
* @throws Exception * @throws Exception
System/Process/SProcess.php
+12 -4
View File
@@ -8,10 +8,18 @@ interface SProcess
{ {
// public function getProcessName(): string; /**
// * @return string
// *
// public function before(\Swoole\Process $process): void; * return process name
*/
public function getProcessName(): string;
/**
* @param \Swoole\Process $process
*/
public function before(\Swoole\Process $process): void;
/** /**
* @param \Swoole\Process $process * @param \Swoole\Process $process
System/Process/ServerInotify.php
+279 -261
View File
@@ -12,10 +12,8 @@ namespace Snowflake\Process;
use Exception; use Exception;
use Snowflake\Abstracts\Config; use Snowflake\Abstracts\Config;
use Snowflake\Exception\ComponentException;
use Snowflake\Exception\ConfigException; use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake; use Snowflake\Snowflake;
use Swoole\Coroutine;
use Swoole\Event; use Swoole\Event;
use Swoole\Timer; use Swoole\Timer;
@@ -25,290 +23,310 @@ use Swoole\Timer;
*/ */
class ServerInotify extends Process class ServerInotify extends Process
{ {
private mixed $inotify; private mixed $inotify;
private bool $isReloading = false; private bool $isReloading = false;
private bool $isReloadingOut = false; private bool $isReloadingOut = false;
private array $watchFiles = []; private array $watchFiles = [];
private ?array $dirs = []; private ?array $dirs = [];
private int $events; private int $events;
private int $int = -1; private int $int = -1;
/** /**
* @param \Swoole\Process $process * @param \Swoole\Process $process
* @throws Exception * @throws ConfigException
*/ */
public function onHandler(\Swoole\Process $process): void public function before(\Swoole\Process $process): void
{ {
set_error_handler([$this, 'onErrorHandler']); // TODO: Implement before() method.
$this->dirs = Config::get('inotify', [APP_PATH]); set_error_handler([$this, 'onErrorHandler']);
if (extension_loaded('inotify')) { $this->dirs = Config::get('inotify', [APP_PATH]);
$this->inotify = inotify_init(); if (extension_loaded('inotify')) {
$this->events = IN_MODIFY | IN_DELETE | IN_CREATE | IN_MOVE; $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 bool $isReload * @return string
* @throws Exception */
*/ public function getProcessName(): string
private function loadDirs($isReload = false) {
{ return 'file change process';
foreach ($this->dirs as $value) { }
if (is_bool($path = realpath($value))) {
continue;
}
if (!is_dir($path)) continue;
$this->loadByDir($path, $isReload);
}
}
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 * @param bool $isReload
*/ * @throws Exception
public function tick() */
{ private function loadDirs($isReload = false)
if ($this->isReloading) { {
return; 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);
} }
}
/** private array $md5Map = [];
* @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;
}
}
}
}
/** /**
* @param $value * @throws Exception
* @param $isReload */
* @return bool public function tick()
*/ {
private function checkFile($value, $isReload): bool if ($this->isReloading) {
{ return;
$md5 = md5($value); }
$mTime = filectime($value);
if (!isset($this->md5Map[$md5])) { $this->loadDirs(true);
if ($isReload) {
return true; Timer::after(2000, [$this, 'tick']);
} }
$this->md5Map[$md5] = $mTime;
} else {
if ($this->md5Map[$md5] != $mTime) {
if ($isReload) {
return true;
}
$this->md5Map[$md5] = $mTime;
}
}
return false;
}
/** /**
* 开始监听 * @param $path
*/ * @param bool $isReload
public function check() * @return void
{ * @throws Exception
if (!($events = inotify_read($this->inotify))) { */
return; private function loadByDir($path, $isReload = false): void
} {
if ($this->isReloading) { if (!is_string($path)) {
if (!$this->isReloadingOut) { return;
$this->isReloadingOut = true; }
} $path = rtrim($path, '/');
return; foreach (glob(realpath($path) . '/*') as $value) {
} if (is_dir($value)) {
$this->loadByDir($value, $isReload);
$eventList = [IN_CREATE, IN_DELETE, IN_MODIFY, IN_MOVED_TO, IN_MOVED_FROM]; }
foreach ($events as $ev) { if (is_file($value)) {
if (empty($ev['name'])) { if ($this->checkFile($value, $isReload)) {
continue; $this->timerReload();
} break;
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 $value
* @throws Exception * @param $isReload
*/ * @return bool
public function trigger_reload() */
{ private function checkFile($value, $isReload): bool
exec(PHP_BINARY . ' snowflake runtime:builder'); {
$md5 = md5($value);
Snowflake::reload(); $mTime = filectime($value);
} if (!isset($this->md5Map[$md5])) {
if ($isReload) {
/** return true;
* @throws Exception }
*/ $this->md5Map[$md5] = $mTime;
public function clearWatch() } else {
{ if ($this->md5Map[$md5] != $mTime) {
foreach ($this->watchFiles as $wd) { if ($isReload) {
try { return true;
inotify_rm_watch($this->inotify, $wd); }
} catch (\Throwable $exception) { $this->md5Map[$md5] = $mTime;
logger()->addError($exception, 'throwable'); }
} }
} return false;
$this->watchFiles = []; }
}
/** /**
* @param $code * 开始监听
* @param $message */
* @param $file public function check()
* @param $line {
* @throws Exception if (!($events = inotify_read($this->inotify))) {
*/ return;
protected function onErrorHandler($code, $message, $file, $line) }
{ if ($this->isReloading) {
if (str_contains($message, 'The file descriptor is not an inotify instance')) { if (!$this->isReloadingOut) {
return; $this->isReloadingOut = true;
} }
logger()->debug('Error:' . $message); return;
logger()->debug($file . ':' . $line); }
}
$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
* @throws Exception */
*/ public function trigger_reload()
public function watch($dir): bool {
{ exec(PHP_BINARY . ' snowflake runtime:builder');
//目录不存在
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'])) { Snowflake::reload();
return FALSE; }
}
$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') { * @param $code
$wd = @inotify_add_watch($this->inotify, $path, $this->events); * @param $message
$this->watchFiles[$path] = $wd; * @param $file
} * @param $line
} * @throws Exception
return TRUE; */
} 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;
}
} }