From 5eaf38b952245fc02e51e9ca8520f118e9c4eeb6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Fri, 9 Oct 2020 16:34:33 +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

---
 HttpServer/Action.php       |   4 +-
 Kafka/ConsumerInterface.php |  29 +++++++++
 Kafka/Kafka.php             |  62 ++++++++++++++++++++
 Kafka/KafkaProviders.php    |  52 ++++++++++++++++
 Kafka/Logger.php            | 114 ++++++++++++++++++++++++++++++++++++
 composer.json               |   3 +-
 6 files changed, 261 insertions(+), 3 deletions(-)
 create mode 100644 Kafka/ConsumerInterface.php
 create mode 100644 Kafka/Kafka.php
 create mode 100644 Kafka/KafkaProviders.php
 create mode 100644 Kafka/Logger.php

diff --git a/HttpServer/Action.php b/HttpServer/Action.php
index 10e06cce..ce0c72bf 100644
--- a/HttpServer/Action.php
+++ b/HttpServer/Action.php
@@ -22,7 +22,7 @@ trait Action
 	 * @return mixed
 	 * @throws Exception
 	 */
-	public function restart($socket)
+	public function restart(\HttpServer\Server $socket)
 	{
 		$this->_shutdown($socket);
 
@@ -34,7 +34,7 @@ trait Action
 	 * @param \HttpServer\Server $socket
 	 * @throws Exception
 	 */
-	public function stop($socket)
+	public function stop(\HttpServer\Server $socket)
 	{
 		$this->_shutdown($socket);
 	}
diff --git a/Kafka/ConsumerInterface.php b/Kafka/ConsumerInterface.php
new file mode 100644
index 00000000..477c91d3
--- /dev/null
+++ b/Kafka/ConsumerInterface.php
@@ -0,0 +1,29 @@
+<?php
+
+
+namespace Kafka;
+
+
+/**
+ * Interface ConsumerInterface
+ * @package App\Kafka
+ */
+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
+	 * @return mixed
+	 */
+	public function onHandler(int $offset, $part, int $crc, int $magic, int $attr, int $timestamp, string $key, string $value);
+
+
+}
diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
new file mode 100644
index 00000000..6be4f6d7
--- /dev/null
+++ b/Kafka/Kafka.php
@@ -0,0 +1,62 @@
+<?php
+
+
+namespace Kafka;
+
+
+use Snowflake\Exception\ConfigException;
+use Snowflake\Snowflake;
+use Swoole\Process;
+use Snowflake\Abstracts\Config as SConfig;
+
+/**
+ * Class Queue
+ * @package Queue
+ */
+class Kafka extends \Snowflake\Process\Process
+{
+
+	/**
+	 * @param Process $process
+	 * @throws ConfigException
+	 */
+	public function onHandler(Process $process)
+	{
+		$kafka = SConfig::get('kafka');
+		$config = \Kafka\ConsumerConfig::getInstance();
+		$config->setMetadataRefreshIntervalMs(
+			$kafka['metadataRefreshIntervalMs'] ?? 1000
+		);
+		$config->setMetadataBrokerList($kafka['brokers']);
+		$config->setGroupId($kafka['groupId']);
+		$config->setBrokerVersion($kafka['version']);
+		$config->setTopics($kafka['topics']);
+
+		$consumer = new \Kafka\Consumer();
+		$consumer->setLogger(new Logger());
+		$consumer->start(function ($topic, $part, $message) {
+			$namespace = 'App\Kafka\\' . ucfirst($topic);
+			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());
+			}
+		});
+
+	}
+
+}
diff --git a/Kafka/KafkaProviders.php b/Kafka/KafkaProviders.php
new file mode 100644
index 00000000..7042b8ab
--- /dev/null
+++ b/Kafka/KafkaProviders.php
@@ -0,0 +1,52 @@
+<?php
+
+
+namespace Kafka;
+
+
+use Exception;
+use HttpServer\Server;
+use Snowflake\Abstracts\Config as SConfig;
+use Snowflake\Abstracts\Providers;
+use Snowflake\Application;
+
+
+/**
+ * Class QueueProviders
+ * @package Queue
+ */
+class KafkaProviders extends Providers
+{
+
+	/**
+	 * @param Application $application
+	 * @throws Exception
+	 */
+	public function onImport(Application $application)
+	{
+		/** @var Server $server */
+		$server = $application->get('server');
+
+		$kafka = SConfig::get('kafka');
+		if (empty($kafka) || !($kafka['enable'] ?? false)) {
+			return;
+		}
+		if (!isset($kafka['topics']) || empty($kafka['topics'])) {
+			throw new Exception('kafka configure error.');
+		}
+		if (!isset($kafka['version']) || empty($kafka['version'])) {
+			throw new Exception('kafka configure error.');
+		}
+		if (!isset($kafka['brokers']) || empty($kafka['brokers'])) {
+			throw new Exception('kafka configure error.');
+		}
+		if (!isset($kafka['groupId']) || empty($kafka['groupId'])) {
+			throw new Exception('kafka configure error.');
+		}
+		if (!is_array($kafka['topics'])) {
+			throw new Exception('Add kafka topics must is array.');
+		}
+		$server->addProcess('kafka', Kafka::class);
+	}
+
+}
diff --git a/Kafka/Logger.php b/Kafka/Logger.php
new file mode 100644
index 00000000..6e2e2b85
--- /dev/null
+++ b/Kafka/Logger.php
@@ -0,0 +1,114 @@
+<?php
+
+
+namespace Kafka;
+
+
+use Psr\Log\LoggerInterface;
+use Snowflake\Exception\ComponentException;
+use Snowflake\Snowflake;
+
+/**
+ * Class Logger
+ * @package Kafka
+ */
+class Logger implements LoggerInterface
+{
+
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 */
+    public function emergency($message, array $context = array())
+    {
+        // TODO: Implement emergency() method.
+        var_dump(func_get_args());
+    }
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function alert($message, array $context = array())
+    {
+	    $logger = Snowflake::app()->getLogger();
+	    $logger->debug($message);
+    }
+
+    public function critical($message, array $context = array())
+    {
+        // TODO: Implement critical() method.
+        var_dump(func_get_args());
+    }
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function error($message, array $context = array())
+    {
+	    $logger = Snowflake::app()->getLogger();
+	    $logger->error($message);
+    }
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function warning($message, array $context = array())
+    {
+	    $logger = Snowflake::app()->getLogger();
+	    $logger->warning($message);
+    }
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function notice($message, array $context = array())
+    {
+	    $logger = Snowflake::app()->getLogger();
+	    $logger->info($message);
+    }
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function info($message, array $context = array())
+    {
+	    $logger = Snowflake::app()->getLogger();
+	    $logger->info($message);
+    }
+
+	/**
+	 * @param string $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function debug($message, array $context = array())
+    {
+	    $logger = Snowflake::app()->getLogger();
+	    $logger->debug($message);
+    }
+
+	/**
+	 * @param $level
+	 * @param $message
+	 * @param array $context
+	 * @throws ComponentException
+	 */
+    public function log($level, $message, array $context = array())
+    {
+        $logger = Snowflake::app()->getLogger();
+        $logger->debug($message);
+    }
+
+
+}
diff --git a/composer.json b/composer.json
index 6182ae08..7fd81a07 100644
--- a/composer.json
+++ b/composer.json
@@ -25,7 +25,8 @@
     "ext-inotify": "*",
     "ext-curl": "*",
     "ext-openssl": "*",
-    "amphp/amp": "v1.2.2"
+    "amphp/amp": "v1.2.2",
+    "psr/log": "1.*"
   },
   "autoload": {
     "psr-4": {