From 69be5ec3d7309e15a7b91a3caf01c28a679d8ffb Mon Sep 17 00:00:00 2001
From: lxy <237809796@qq.com>
Date: Fri, 22 Jan 2021 17:53:38 +0800
Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=EF=BC=9A=E8=AE=A2=E9=98=85?=
 =?UTF-8?q?=20subscribe=20=E6=96=B9=E6=B3=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/com/zdemo/IConsumer.java           |  5 +++++
 src/com/zdemo/kafak/KafakConsumer.java | 12 ++++++++++++
 src/com/zdemo/redis/RedisConsumer.java | 12 ++++++++++++
 src/com/zdemo/zdb/ZHubClient.java      | 12 ++++++++++++
 test/com/zdemo/test/AppTest.java       |  3 ---
 5 files changed, 41 insertions(+), 3 deletions(-)

diff --git a/src/com/zdemo/IConsumer.java b/src/com/zdemo/IConsumer.java
index 9353060..011f4e7 100644
--- a/src/com/zdemo/IConsumer.java
+++ b/src/com/zdemo/IConsumer.java
@@ -3,6 +3,7 @@ package com.zdemo;
 import org.redkale.util.TypeToken;
 
 import java.util.Collection;
+import java.util.function.Consumer;
 
 public interface IConsumer {
     TypeToken<String> TYPE_TOKEN_STRING = new TypeToken<String>() {
@@ -22,4 +23,8 @@ public interface IConsumer {
      * @param topic
      */
     void unsubscribe(String topic);
+
+    void subscribe(String topic, Consumer<String> consumer);
+    
+    <T> void subscribe(String topic, TypeToken<T> typeToken, Consumer<T> consumer);
 }
diff --git a/src/com/zdemo/kafak/KafakConsumer.java b/src/com/zdemo/kafak/KafakConsumer.java
index c87ff9a..1ceca8d 100644
--- a/src/com/zdemo/kafak/KafakConsumer.java
+++ b/src/com/zdemo/kafak/KafakConsumer.java
@@ -9,6 +9,7 @@ import org.apache.kafka.common.errors.WakeupException;
 import org.redkale.net.http.RestService;
 import org.redkale.service.Service;
 import org.redkale.util.AnyValue;
+import org.redkale.util.TypeToken;
 
 import javax.annotation.Resource;
 import java.io.File;
@@ -17,6 +18,7 @@ import java.io.IOException;
 import java.time.Duration;
 import java.util.Properties;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -99,4 +101,14 @@ public abstract class KafakConsumer extends AbstractConsumer implements IConsume
     public void unsubscribe(String topic) {
         queue.add(() -> eventMap.remove(topic)); // 加入延时执行队列（下一次订阅变更检查周期执行）
     }
+
+    @Override
+    public void subscribe(String topic, Consumer<String> consumer) {
+        addEventType(EventType.of(topic, consumer));
+    }
+
+    @Override
+    public <T> void subscribe(String topic, TypeToken<T> typeToken, Consumer<T> consumer) {
+        addEventType(EventType.of(topic, typeToken, consumer));
+    }
 }
diff --git a/src/com/zdemo/redis/RedisConsumer.java b/src/com/zdemo/redis/RedisConsumer.java
index 6831133..1672973 100644
--- a/src/com/zdemo/redis/RedisConsumer.java
+++ b/src/com/zdemo/redis/RedisConsumer.java
@@ -5,6 +5,7 @@ import com.zdemo.EventType;
 import com.zdemo.IConsumer;
 import org.redkale.service.Service;
 import org.redkale.util.AnyValue;
+import org.redkale.util.TypeToken;
 
 import javax.annotation.Resource;
 import java.io.BufferedReader;
@@ -13,6 +14,7 @@ import java.io.InputStreamReader;
 import java.io.OutputStreamWriter;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -119,4 +121,14 @@ public class RedisConsumer extends AbstractConsumer implements IConsumer, Servic
             logger.log(Level.WARNING, "", e);
         }
     }
+
+    @Override
+    public void subscribe(String topic, Consumer<String> consumer) {
+        addEventType(EventType.of(topic, consumer));
+    }
+
+    @Override
+    public <T> void subscribe(String topic, TypeToken<T> typeToken, Consumer<T> consumer) {
+        addEventType(EventType.of(topic, typeToken, consumer));
+    }
 }
diff --git a/src/com/zdemo/zdb/ZHubClient.java b/src/com/zdemo/zdb/ZHubClient.java
index 84bed93..d62b97a 100644
--- a/src/com/zdemo/zdb/ZHubClient.java
+++ b/src/com/zdemo/zdb/ZHubClient.java
@@ -4,6 +4,7 @@ import com.zdemo.*;
 import org.redkale.convert.json.JsonConvert;
 import org.redkale.service.Service;
 import org.redkale.util.AnyValue;
+import org.redkale.util.TypeToken;
 
 import javax.annotation.Resource;
 import java.io.BufferedReader;
@@ -15,6 +16,7 @@ import java.net.Socket;
 import java.net.SocketException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -219,4 +221,14 @@ public abstract class ZHubClient extends AbstractConsumer implements IConsumer,
             this.runnable = runnable;
         }
     }
+
+    @Override
+    public void subscribe(String topic, Consumer<String> consumer) {
+        addEventType(EventType.of(topic, consumer));
+    }
+
+    @Override
+    public <T> void subscribe(String topic, TypeToken<T> typeToken, Consumer<T> consumer) {
+        addEventType(EventType.of(topic, typeToken, consumer));
+    }
 }
diff --git a/test/com/zdemo/test/AppTest.java b/test/com/zdemo/test/AppTest.java
index 5affae8..b60d621 100644
--- a/test/com/zdemo/test/AppTest.java
+++ b/test/com/zdemo/test/AppTest.java
@@ -9,13 +9,10 @@ import org.redkale.util.Utility;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
-import static java.util.Arrays.asList;
-
 /**
  * 消息发布订阅测试
  */