diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 9290d4d8..11a674df 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -35,27 +35,28 @@ class Kafka extends \Snowflake\Process\Process $consumer = new \Kafka\Consumer(); $consumer->setLogger(new Logger()); $consumer->start(function ($topic, $part, $message) { - var_dump($topic, $part, $message); -// $namespace = 'App\Kafka\\' . ucfirst($topic) . 'Consumer'; -// if (!class_exists($namespace) || !($namespace instanceof ConsumerInterface)) { -// return; -// } -// try { -// /** @var ConsumerInterface $class */ -// $class = Snowflake::createObject($namespace); -// $class->onHandler( -// $message['offset'], -// $part, -// $message['message']['crc'], -// $message['message']['magic'], -// $message['message']['attr'], -// $message['message']['timestamp'], -// $message['message']['key'], -// $message['message']['value'] -// ); -// } catch (\Throwable $exception) { -// $this->application->error($exception->getMessage()); -// } + try { + $namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer'; + if (!class_exists($namespace)) { + return; + } + if ($namespace instanceof ConsumerInterface) { + /** @var ConsumerInterface $class */ + $class = Snowflake::createObject($namespace); + $class->onHandler( + $message['offset'], + $part, + $message['message']['crc'], + $message['message']['magic'], + $message['message']['attr'], + $message['message']['timestamp'], + $message['message']['key'], + $message['message']['value'] + ); + } + } catch (\Throwable $exception) { + $this->application->error($exception->getMessage()); + } }); }