diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index aa16d6cf..37198209 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -73,7 +73,7 @@ class Kafka extends CustomProcess $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); do { - if ($this->isExit()) { + if ($this->checkProcessIsStop()) { $this->exit(); break; } diff --git a/Server/Abstracts/CustomProcess.php b/Server/Abstracts/CustomProcess.php index fe30bb0e..bcb510d7 100644 --- a/Server/Abstracts/CustomProcess.php +++ b/Server/Abstracts/CustomProcess.php @@ -17,8 +17,25 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess protected bool $enableSwooleCoroutine = true; - /** @var Coroutine\Channel */ - protected Coroutine\Channel $channel; + protected bool $isStop = false; + + + /** + * + */ + public function onProcessStop(): void + { + $this->isStop = true; + } + + + /** + * @return bool + */ + public function checkProcessIsStop(): bool + { + return $this->isStop === true; + } /** @@ -26,22 +43,17 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess */ public function signListen(Process $process): void { - $this->channel = new Coroutine\Channel(1); - $this->channel->push(1); - if (!$this->enableSwooleCoroutine) { Process::signal(SIGTERM | SIGKILL, function ($signo) use ($process) { - putenv('processStatus=exit'); - + $this->onProcessStop(); $this->waiteExit($process); }); } else { go(function () use ($process) { $data = Coroutine::waitSignal(SIGTERM | SIGKILL, -1); if ($data) { - putenv('processStatus=exit'); - + $this->onProcessStop(); $this->waiteExit($process); } }); @@ -49,40 +61,21 @@ abstract class CustomProcess implements \Server\SInterface\CustomProcess } - /** - * @return string - */ - #[Pure] protected function getStatus(): string - { - return env('processStatus', 'working'); - } - - - /** - * @return bool - */ - #[Pure] protected function isExit(): bool - { - return $this->getStatus() == 'exit'; - } - - /** * */ - protected function exit() + protected function exit(): void { - $this->channel->pop(); - $this->channel->close(); + putenv('process.status=idle'); } /** * @return bool */ - public function isWorking(): bool + #[Pure] public function isWorking(): bool { - return $this->channel->isEmpty(); + return env('process.status', 'working') == 'working'; } diff --git a/Server/SInterface/CustomProcess.php b/Server/SInterface/CustomProcess.php index 91f9c1a5..32e8a8e5 100644 --- a/Server/SInterface/CustomProcess.php +++ b/Server/SInterface/CustomProcess.php @@ -34,4 +34,10 @@ interface CustomProcess public function onHandler(Process $process): void; + /** + * + */ + public function onProcessStop(): void; + + } diff --git a/System/Crontab/Zookeeper.php b/System/Crontab/Zookeeper.php index deba7f55..5bd8d659 100644 --- a/System/Crontab/Zookeeper.php +++ b/System/Crontab/Zookeeper.php @@ -60,9 +60,9 @@ class Zookeeper extends CustomProcess */ public function loop() { - if ($this->isExit()) { - Timer::clear($this->_timer); + if ($this->checkProcessIsStop()) { $this->exit(); + Timer::clear($this->_timer); return; } $redis = Kiri::app()->getRedis(); diff --git a/System/Error/LoggerProcess.php b/System/Error/LoggerProcess.php index c997f61c..7ef8356a 100644 --- a/System/Error/LoggerProcess.php +++ b/System/Error/LoggerProcess.php @@ -49,7 +49,7 @@ class LoggerProcess extends CustomProcess */ public function message(Process $process) { - if ($this->isExit()) { + if ($this->checkProcessIsStop()) { $this->exit(); return; }