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-15 18:18:41 +08:00
parent 47c1ab2086
commit bd57ba367a
2 changed files with 103 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HttpServer/Events/OnPipeMessage.php
+36 -33
View File
@@ -9,7 +9,6 @@ use HttpServer\Abstracts\Callback;
use Kafka\Struct;
use Snowflake\Crontab\Crontab;
use Snowflake\Event;
use Snowflake\Exception\ComponentException;
use Swoole\Server;
/**
@@ -25,20 +24,20 @@ class OnPipeMessage extends Callback
* @param $swollen_universalize
* @throws Exception
*/
public function onHandler(Server $server, int $src_worker_id, $swollen_universalize)
{
try {
match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize)
};
} catch (\Throwable $exception) {
$this->addError($exception);
} finally {
fire(Event::SYSTEM_RESOURCE_RELEASES);
}
}
public function onHandler(Server $server, int $src_worker_id, $swollen_universalize)
{
try {
match ($swollen_universalize['action'] ?? null) {
'kafka' => $this->onKafkaWorker($swollen_universalize),
'crontab' => $this->onCrontabWorker($swollen_universalize),
default => $this->onMessageWorker($server, $src_worker_id, $swollen_universalize)
};
} catch (\Throwable $exception) {
$this->addError($exception);
} finally {
fire(Event::SYSTEM_RESOURCE_RELEASES);
}
}
/**
@@ -46,13 +45,17 @@ class OnPipeMessage extends Callback
* @return string
* @throws Exception
*/
private function onCrontabWorker(array $message): string
{
/** @var Crontab $crontab */
$crontab = $message['handler'] ?? null;
$crontab->increment()->execute();
return 'success';
}
private function onCrontabWorker(array $message): string
{
/** @var Crontab $crontab */
$crontab = $message['handler'] ?? null;
if (!isset($message['handler'])) {
throw new Exception('unknown handler');
}
$crontab = swoole_unserialize($crontab['handler']);
$crontab->increment()->execute();
return 'success';
}
/**
@@ -62,26 +65,26 @@ class OnPipeMessage extends Callback
* @return string
* @throws Exception
*/
private function onMessageWorker($server, $src_worker_id, $message): string
{
fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]);
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'];
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';
}
}