zhangdeman/event
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

redis驱动发送事件

Browse Source
This commit is contained in:
白茶清欢
2024-06-27 11:50:13 +08:00
parent 6f56ff3c4f
commit 414ad47e72
6 changed files with 95 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
redis_pub_sub.go
View File

@ -14,6 +14,7 @@ import (
"git.zhangdeman.cn/zhangdeman/consts"
"git.zhangdeman.cn/zhangdeman/event/abstract"
"git.zhangdeman.cn/zhangdeman/event/define"
"git.zhangdeman.cn/zhangdeman/wrapper"
"github.com/redis/go-redis/v9"
"time"
)
@ -50,14 +51,58 @@ type RedisEventPubSub struct {
panicCallback abstract.PanicCallback // panic回调
}
func (r *RedisEventPubSub) SendEvent(ctx context.Context, eventData *define.EventData, sendSuccessCallback abstract.SendSuccessCallback, sendFailCallback abstract.SendFailCallback) (*define.SendResult, error) {
//TODO implement me
panic("implement me")
// SendEvent 发布时间
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 10:58 2024/6/27
func (r *RedisEventPubSub) SendEvent(ctx context.Context, eventData *define.EventData) (*define.SendResult, error) {
partition := int(wrapper.String(eventData.Key).HashNumber().Value % uint64(r.pubSubConfig.PartitionNum))
realTopic := fmt.Sprintf("%v_%v", r.pubSubConfig.Topic, partition)
res := r.redisClient.Publish(ctx, realTopic, eventData)
if nil == res {
return nil, errors.New("can not get send event result")
}
return &define.SendResult{
Data: eventData,
Partition: partition,
Topic: realTopic,
IsSuccess: res.Err() == nil,
FailReason: wrapper.TernaryOperator.String(res.Err() == nil, "success", wrapper.String(res.Err().Error())).Value(),
Extension: nil,
}, res.Err()
}
// SendEventAsync 异步发送事件
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 11:29 2024/6/27
func (r *RedisEventPubSub) SendEventAsync(ctx context.Context, eventData *define.EventData, sendSuccessCallback abstract.SendSuccessCallback, sendFailCallback abstract.SendFailCallback) {
//TODO implement me
panic("implement me")
go func() {
if nil == sendFailCallback {
sendFailCallback = define.DefaultSendFailCallback
}
if nil == sendSuccessCallback {
sendSuccessCallback = define.DefaultSendSuccessCallback
}
var (
sendResult *define.SendResult
handlerErr error
)
defer func() {
if panicErr := recover(); nil != panicErr {
r.panicCallback(panicErr, eventData, map[string]any{
"send_result": sendResult,
})
}
}()
if sendResult, handlerErr = r.SendEvent(ctx, eventData); nil != handlerErr {
sendFailCallback(ctx, eventData, sendResult, handlerErr)
} else {
sendSuccessCallback(ctx, sendResult)
}
}()
}
// GetConsumeEventChan 获取消息channel, 自行实现消费