diff --git a/Kafka/ConsumerInterface.php b/Kafka/ConsumerInterface.php index 477c91d3..c9437f80 100644 --- a/Kafka/ConsumerInterface.php +++ b/Kafka/ConsumerInterface.php @@ -13,17 +13,10 @@ interface ConsumerInterface /** - * @param int $offset - * @param $part - * @param int $crc - * @param int $magic - * @param int $attr - * @param int $timestamp - * @param string $key - * @param string $value + * @param Struct $struct * @return mixed */ - public function onHandler(int $offset, $part, int $crc, int $magic, int $attr, int $timestamp, string $key, string $value); + public function onHandler(Struct $struct); } diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 6fd33723..872d5e3e 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -43,16 +43,7 @@ class Kafka extends \Snowflake\Process\Process } $class = Snowflake::createObject($namespace); if ($class instanceof ConsumerInterface) { - $class->onHandler( - $message['offset'], - $part, - $message['message']['crc'], - $message['message']['magic'], - $message['message']['attr'], - $message['message']['timestamp'], - $message['message']['key'], - $message['message']['value'] - ); + $class->onHandler(new Struct($topic, $part, $message)); } } catch (\Throwable $exception) { $this->application->error($exception->getMessage()); diff --git a/Kafka/Struct.php b/Kafka/Struct.php new file mode 100644 index 00000000..60cc5020 --- /dev/null +++ b/Kafka/Struct.php @@ -0,0 +1,42 @@ +topic = $topic; + $this->offset = $message['offset']; + $this->part = $part; + $this->crc = $message['message']['crc']; + $this->magic = $message['message']['magic']; + $this->attr = $message['message']['attr']; + $this->timestamp = $message['message']['timestamp']; + $this->key = $message['message']['key']; + $this->value = $message['message']['value']; + } + +}