From aa664d4181eb5f4a32a4a31a26f7f21f4d2c71fa Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Thu, 14 Jan 2021 10:41:16 +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 | 38 ++++++++++++++++++++++----------------
 1 file changed, 22 insertions(+), 16 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index a128b9d4..8eb354a1 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -18,6 +18,7 @@ use Swoole\Coroutine\WaitGroup;
 use Swoole\IDEHelper\StubGenerators\Swoole;
 use Swoole\Process;
 use Snowflake\Abstracts\Config as SConfig;
+use Swoole\Timer;
 use Throwable;
 
 /**
@@ -39,6 +40,9 @@ class Kafka extends \Snowflake\Process\Process
 		$this->channelListener();
 
 		$this->waite(json_decode($process->read(), true));
+		Timer::tick(1000, function () {
+			$this->application->error('kafka service error.');
+		});
 	}
 
 
@@ -49,6 +53,9 @@ class Kafka extends \Snowflake\Process\Process
 	{
 		try {
 			[$config, $topic, $conf] = $this->kafkaConfig($kafkaServer);
+			if (empty($config) && empty($topic) && empty($conf)) {
+				return;
+			}
 			$objRdKafka = new Consumer($config);
 			$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
 
@@ -144,19 +151,19 @@ class Kafka extends \Snowflake\Process\Process
 	 */
 	private function kafkaConfig($kafka): array
 	{
-		$conf = new Conf();
-		$conf->setRebalanceCb([$this, 'rebalanced_cb']);
-		$conf->set('group.id', $kafka['groupId']);
-		$conf->set('metadata.broker.list', $kafka['brokers']);
-		$conf->set('socket.timeout.ms', '30000');
-
-		if (function_exists('pcntl_sigprocmask')) {
-			pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
-			$conf->set('internal.termination.signal', (string)SIGIO);
-		}
-
-		$topicConf = new TopicConf();
 		try {
+			$conf = new Conf();
+			$conf->setRebalanceCb([$this, 'rebalanced_cb']);
+			$conf->set('group.id', $kafka['groupId']);
+			$conf->set('metadata.broker.list', $kafka['brokers']);
+			$conf->set('socket.timeout.ms', '30000');
+
+			if (function_exists('pcntl_sigprocmask')) {
+				pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
+				$conf->set('internal.termination.signal', (string)SIGIO);
+			}
+
+			$topicConf = new TopicConf();
 			$topicConf->set('auto.commit.enable', '1');
 			$topicConf->set('auto.commit.interval.ms', '100');
 
@@ -166,14 +173,13 @@ class Kafka extends \Snowflake\Process\Process
 			$topicConf->set('offset.store.path', 'kafka_offset.log');
 			$topicConf->set('offset.store.method', 'broker');
 
-			var_dump(get_object_vars($topicConf));
-
-			$topicConf->set('auto.create.topics.enable', 'true');
+			return [$conf, $topicConf, $kafka];
 		} catch (Throwable $exception) {
 			$this->application->error($exception->getMessage());
+
+			return [null, null, null];
 		}
 
-		return [$conf, $topicConf, $kafka];
 	}