改名
This commit is contained in:
+15
-13
@@ -118,21 +118,23 @@ class Kafka extends \Snowflake\Process\Process
|
|||||||
*/
|
*/
|
||||||
protected function handlerExecute($topic, $message)
|
protected function handlerExecute($topic, $message)
|
||||||
{
|
{
|
||||||
try {
|
go(function () use ($topic, $message) {
|
||||||
$topic = str_replace('-', '_', $topic);
|
try {
|
||||||
|
$topic = str_replace('-', '_', $topic);
|
||||||
|
|
||||||
$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
|
$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
|
||||||
if (!class_exists($namespace)) {
|
if (!class_exists($namespace)) {
|
||||||
return;
|
return;
|
||||||
|
}
|
||||||
|
$class = Snowflake::createObject($namespace);
|
||||||
|
if (!($class instanceof ConsumerInterface)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
$class->onHandler(new Struct($topic, $message));
|
||||||
|
} catch (Throwable $exception) {
|
||||||
|
$this->application->error($exception->getMessage());
|
||||||
}
|
}
|
||||||
$class = Snowflake::createObject($namespace);
|
});
|
||||||
if (!($class instanceof ConsumerInterface)) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
$class->onHandler(new Struct($topic, $message));
|
|
||||||
} catch (Throwable $exception) {
|
|
||||||
$this->application->error($exception->getMessage());
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user