event/redis_pub_sub.go

// Package event ...
//
// Description : event ...
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 2024-06-25 16:06
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 (
	RedisEventPubSubClient abstract.IEvent
)

// InitRedisPubSubEvent 初始化redis事件驱动, 基于 pub / sub 指令
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 16:24 2024/6/25
func InitRedisPubSubEvent(redisClient *redis.Client, pubSubConfig *define.RedisEventPubSubConfig) {
	instance := &RedisEventPubSub{
		redisClient:  redisClient,
		pubSubConfig: pubSubConfig,
	}
	instance.SetRedisClient(redisClient, pubSubConfig)
	RedisEventPubSubClient = instance
}

// RedisEventPubSub 基于redis的事件驱动
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 16:07 2024/6/25
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) {
	//TODO implement me
	panic("implement me")
}

func (r *RedisEventPubSub) SendEventAsync(ctx context.Context, eventData *define.EventData) {
	//TODO implement me
	panic("implement me")
}

func (r *RedisEventPubSub) SendFailCallback(ctx context.Context, eventResult *define.SendResult) {
	//TODO implement me
	panic("implement me")
}

func (r *RedisEventPubSub) SendSuccessCallback(ctx context.Context, eventResult *define.SendResult, err error) {
	//TODO implement me
	panic("implement me")
}

func (r *RedisEventPubSub) ConsumeEvent() (<-chan *define.EventData, error) {
	//TODO implement me
	panic("implement me")
}

func (r *RedisEventPubSub) ConsumeFailCallback(eventData *define.EventData, err error) {
	//TODO implement me
	panic("implement me")
}

func (r *RedisEventPubSub) ConsumeSuccessCallback(eventData *define.EventData) {
	//TODO implement me
	panic("implement me")
}

// Destroy 销毁事件实例
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:42 2024/6/26
func (r *RedisEventPubSub) Destroy() {
	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 {
	return consts.EventDriverRedis
}

// SetRedisClient 设置redis客户端
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 16:30 2024/6/25
func (r *RedisEventPubSub) SetRedisClient(redisClient *redis.Client, pubSubConfig *define.RedisEventPubSubConfig) {
	if nil == pubSubConfig {
		pubSubConfig = &define.RedisEventPubSubConfig{
			Topic:             "",
			PartitionNum:      0,
			MessageBufferSize: 0,
		}
	}
	if len(pubSubConfig.Topic) == 0 {
		pubSubConfig.Topic = define.DefaultRedisTopic
	}
	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)
}