From 87b2742e360a4c5bf17dfbfc8bf4e026b5e49317 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=99=BD=E8=8C=B6=E6=B8=85=E6=AC=A2?=
 <go_developer@163.com>
Date: Wed, 26 Jun 2024 15:46:00 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=9F=BA=E4=BA=8Eredis?=
 =?UTF-8?q?=E7=9A=84=E4=BA=8B=E4=BB=B6=E9=94=80=E6=AF=81=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 define/redis.go  | 12 ++++++++----
 redis_pub_sub.go | 45 +++++++++++++++++++++++++++++++++++++++------
 2 files changed, 47 insertions(+), 10 deletions(-)

diff --git a/define/redis.go b/define/redis.go
index 5892399..60dfcc9 100644
--- a/define/redis.go
+++ b/define/redis.go
@@ -8,8 +8,10 @@
 package define
 
 const (
-	DefaultRedisPartitionNum = 1
-	DefaultRedisTopic        = "EVENT_TOPIC_C6DBE0AAE846C5C0DE35802107326B1E"
+	DefaultRedisPartitionNum      = 1
+	DefaultRedisTopic             = "EVENT_TOPIC_C6DBE0AAE846C5C0DE35802107326B1E"
+	DefaultRedisMessageBufferSize = 1024
+	DefaultRedisCloseMaxWaitTime  = 5000 // 默认最大等待 : 5s
 )
 
 // RedisEventPubSubConfig redis事件配置
@@ -18,6 +20,8 @@ const (
 //
 // Date : 16:25 2024/6/25
 type RedisEventPubSubConfig struct {
-	Topic        string `json:"topic"`         // topic key, 不指定随机生成
-	PartitionNum int    `json:"partition_num"` // 多少个分区, 默认值 : 1
+	Topic             string `json:"topic"`               // topic key, 不指定随机生成
+	PartitionNum      int    `json:"partition_num"`       // 多少个分区, 默认值 : 1
+	MessageBufferSize int    `json:"message_buffer_size"` // 消息缓冲区大小
+	CloseMaxWaitTime  int    `json:"close_max_wait_time"` // 关闭消息实例, 最大等待时长, 单位 : ms
 }
diff --git a/redis_pub_sub.go b/redis_pub_sub.go
index 6db7505..e6eefd2 100644
--- a/redis_pub_sub.go
+++ b/redis_pub_sub.go
@@ -9,9 +9,11 @@ package event
 
 import (
 	"context"
+	"git.zhangdeman.cn/zhangdeman/consts"
 	"git.zhangdeman.cn/zhangdeman/event/abstract"
 	"git.zhangdeman.cn/zhangdeman/event/define"
 	"github.com/redis/go-redis/v9"
+	"time"
 )
 
 var (
@@ -40,6 +42,8 @@ func InitRedisPubSubEvent(redisClient *redis.Client, pubSubConfig *define.RedisE
 type RedisEventPubSub struct {
 	redisClient  *redis.Client                  // redis客户端
 	pubSubConfig *define.RedisEventPubSubConfig // 事件配置
+	messageChan  chan *define.EventData         // 消息队列
+	stopConsumer chan bool                      // 停止消费者
 }
 
 func (r *RedisEventPubSub) SendEvent(ctx context.Context, eventData *define.EventData) (*define.SendResult, error) {
@@ -77,14 +81,34 @@ func (r *RedisEventPubSub) ConsumeSuccessCallback(eventData *define.EventData) {
 	panic("implement me")
 }
 
+// Destroy 销毁事件实例
+//
+// Author : go_developer@163.com<白茶清欢>
+//
+// Date : 15:42 2024/6/26
 func (r *RedisEventPubSub) Destroy() {
-	//TODO implement me
-	panic("implement me")
+	r.stopConsumer <- true // 停止消费者
+	messageChan := make(chan bool, 1)
+	go func() {
+		for {
+			if len(r.messageChan) == 0 {
+				messageChan <- true
+				return
+			}
+		}
+	}()
+	select {
+	case <-time.After(time.Millisecond * time.Duration(r.pubSubConfig.CloseMaxWaitTime)):
+		// 定时器到期
+		return
+	case <-messageChan:
+		// 没有待消费数据
+		return
+	}
 }
 
 func (r *RedisEventPubSub) DriverType() string {
-	//TODO implement me
-	panic("implement me")
+	return consts.EventDriverRedis
 }
 
 // SetRedisClient 设置redis客户端
@@ -95,8 +119,9 @@ func (r *RedisEventPubSub) DriverType() string {
 func (r *RedisEventPubSub) SetRedisClient(redisClient *redis.Client, pubSubConfig *define.RedisEventPubSubConfig) {
 	if nil == pubSubConfig {
 		pubSubConfig = &define.RedisEventPubSubConfig{
-			Topic:        "",
-			PartitionNum: 0,
+			Topic:             "",
+			PartitionNum:      0,
+			MessageBufferSize: 0,
 		}
 	}
 	if len(pubSubConfig.Topic) == 0 {
@@ -105,6 +130,14 @@ func (r *RedisEventPubSub) SetRedisClient(redisClient *redis.Client, pubSubConfi
 	if pubSubConfig.PartitionNum <= 0 {
 		pubSubConfig.PartitionNum = define.DefaultRedisPartitionNum
 	}
+	if pubSubConfig.MessageBufferSize <= 0 {
+		r.pubSubConfig.MessageBufferSize = define.DefaultRedisMessageBufferSize
+	}
+	if pubSubConfig.CloseMaxWaitTime <= 0 {
+		r.pubSubConfig.CloseMaxWaitTime = define.DefaultRedisCloseMaxWaitTime
+	}
 	r.redisClient = redisClient
 	r.pubSubConfig = pubSubConfig
+	r.stopConsumer = make(chan bool, 1)
+	r.messageChan = make(chan *define.EventData, r.pubSubConfig.MessageBufferSize)
 }