From f0eb29fbd55ee8d1d822f65a8de43bce3374df84 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Thu, 5 Aug 2021 17:02:12 +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

---
 Annotation/Kafka.php    |  3 +--
 Kafka/KafkaProvider.php |  1 -
 Kafka/Message.php       | 13 +++++++++----
 3 files changed, 10 insertions(+), 7 deletions(-)

diff --git a/Annotation/Kafka.php b/Annotation/Kafka.php
index a4052a48..f77e5ce5 100644
--- a/Annotation/Kafka.php
+++ b/Annotation/Kafka.php
@@ -35,13 +35,12 @@ use Snowflake\Snowflake;
 	 */
     public function execute(mixed $class, mixed $method = null): bool
     {
-    	var_dump(class_implements($class));
         if (!($class instanceof ConsumerInterface)) {
             return false;
         }
 
         /** @var KafkaProvider $container */
-        $container = Snowflake::app()->get('kafka-container');
+        $container = Snowflake::getDi()->get(KafkaProvider::class);
         $container->addConsumer($this->topic, $class);
 
         return true;
diff --git a/Kafka/KafkaProvider.php b/Kafka/KafkaProvider.php
index eba7882e..b15f0737 100644
--- a/Kafka/KafkaProvider.php
+++ b/Kafka/KafkaProvider.php
@@ -27,7 +27,6 @@ class KafkaProvider extends BaseObject
         if (isset($this->_topics[$topic])) {
             return;
         }
-        var_dump($topic, $handler);
         $this->_topics[$topic] = $handler::class;
     }
 
diff --git a/Kafka/Message.php b/Kafka/Message.php
index 701d0ba0..81e8e2a0 100644
--- a/Kafka/Message.php
+++ b/Kafka/Message.php
@@ -4,11 +4,13 @@ namespace Kafka;
 
 
 use Server\SInterface\PipeMessage;
+use Snowflake\Exception\NotFindClassException;
+use Snowflake\Snowflake;
 
 /**
  *
  */
-class Message implements ConsumerInterface, PipeMessage
+class Message implements PipeMessage
 {
 
 
@@ -21,12 +23,15 @@ class Message implements ConsumerInterface, PipeMessage
 
 
 	/**
-	 *
+	 * @throws \ReflectionException
+	 * @throws NotFindClassException
 	 */
 	public function process(): void
 	{
-		// TODO: Implement process() method.
-		var_dump($this->struct);
+		/** @var KafkaProvider $container */
+		$container = Snowflake::getDi()->get(KafkaProvider::class);
+		$data = $container->getConsumer($this->struct->topic);
+		var_dump($data);
 	}
 
 }