diff --git a/Kafka/ConsumerInterface.php b/Kafka/ConsumerInterface.php index 13dcfca4..fabcac91 100644 --- a/Kafka/ConsumerInterface.php +++ b/Kafka/ConsumerInterface.php @@ -12,13 +12,6 @@ interface ConsumerInterface { - /** - * @param Struct $struct - */ - public function setParams(Struct $struct): void; - - - /** * @return mixed */ diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 7563075a..c2bddbd0 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -124,17 +124,7 @@ class Kafka implements CustomProcess $setting = $server->setting['worker_num']; - $container = Snowflake::app()->get('kafka-container'); - $handler = $container->getConsumer($topic); - - var_dump($container); - if (!empty($handler)) { - /** @var ConsumerInterface $data */ - $data = new $handler(); - $data->setParams(new Struct($topic, $message)); - - $server->sendMessage($data, random_int(0, $setting - 1)); - } + $server->sendMessage(new Message(new Struct($topic, $message)), random_int(0, $setting - 1)); } catch (Throwable $exception) { logger()->addError($exception, 'throwable'); } diff --git a/Kafka/Message.php b/Kafka/Message.php new file mode 100644 index 00000000..e57d1acb --- /dev/null +++ b/Kafka/Message.php @@ -0,0 +1,31 @@ +