From ad0d48f775b84a02a35e6902ea3f39b7f075aaff Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Mon, 26 Oct 2020 17:08:52 +0800
Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Kafka/Kafka.php | 32 ++++++++++++++++++++++++++------
 1 file changed, 26 insertions(+), 6 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index d16b327e..cac37839 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -22,24 +22,44 @@ use function Amp\stop;
 class Kafka extends \Snowflake\Process\Process
 {
 
-	protected Channel $channel;
+	/** @var Channel */
+	protected $channel;
+
+
+	/**
+	 * @throws ConfigException
+	 */
+//	public function initConfig()
+//	{
+//		$kafka = SConfig::get('kafka');
+//		$config = ConsumerConfig::getInstance();
+//		$config->setMetadataRefreshIntervalMs(
+//			$kafka['metadataRefreshIntervalMs'] ?? 1000
+//		);
+//		$config->setMetadataBrokerList($kafka['brokers']);
+//		$config->setGroupId($kafka['groupId']);
+//		$config->setBrokerVersion($kafka['version']);
+//		$config->setTopics($kafka['topics']);
+//		$this->channelListener($kafka);
+//
+//		return [new Consumer(), $kafka];
+//	}
 
 
 	/**
 	 * @param Process $process
 	 * @throws ConfigException
+	 * @throws \RdKafka\Exception
 	 * @throws \Exception
 	 */
 	public function onHandler(Process $process)
 	{
 		$this->channelListener();
 		[$config, $conf] = $this->kafkaConfig();
-		$consumer = new \RdKafka\Consumer($config);
-		$consumer->addBrokers($conf['brokers']);
-
-		$topic = $consumer->newTopic("test");
+		$consumer = new KafkaConsumer($config);
+		$consumer->subscribe($conf['topics']);
 		while (true) {
-			$message = $topic->consume(0, 1000);
+			$message = $consumer->consume($conf['metadataRefreshIntervalMs'] ?? 1000);
 			if (empty($message)) {
 				continue;
 			}