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

增加获取消息channel方法

Browse Source
This commit is contained in:
白茶清欢 2024-07-17 12:55:51 +08:00
parent 10f0eb484f
commit 5fc71de49a
2 changed files with 5 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
define/memory.go
View File

@ -8,7 +8,6 @@
package define package define
const ( const (
DefaultMemoryPartitionNum = 1 // 默认的分区数量
DefaultMemoryChannelSize = 1024 // 默认的消息channel大小 DefaultMemoryChannelSize = 1024 // 默认的消息channel大小
DefaultMemoryCloseMaxWaitTime = 5000 // 默认最大等待 : 5s DefaultMemoryCloseMaxWaitTime = 5000 // 默认最大等待 : 5s
) )
@ -19,7 +18,6 @@ const (
// //
// Date : 11:26 2024/7/17 // Date : 11:26 2024/7/17
type MemoryEventConfig struct { type MemoryEventConfig struct {
PartitionNum int `json:"partition_num"` // 多少个分区, 默认值 : 1
MessageBufferSize int `json:"message_buffer_size"` // 消息缓冲区大小 MessageBufferSize int `json:"message_buffer_size"` // 消息缓冲区大小
CloseMaxWaitTime int `json:"close_max_wait_time"` // 关闭消息实例, 最大等待时长, 单位 : ms CloseMaxWaitTime int `json:"close_max_wait_time"` // 关闭消息实例, 最大等待时长, 单位 : ms
} }

19
memory.go
View File

@ -13,7 +13,6 @@ import (
"git.zhangdeman.cn/zhangdeman/consts" "git.zhangdeman.cn/zhangdeman/consts"
"git.zhangdeman.cn/zhangdeman/event/abstract" "git.zhangdeman.cn/zhangdeman/event/abstract"
"git.zhangdeman.cn/zhangdeman/event/define" "git.zhangdeman.cn/zhangdeman/event/define"
"git.zhangdeman.cn/zhangdeman/wrapper"
) )
var ( var (
@ -44,7 +43,7 @@ func InitMemoryEvent(cfg *define.MemoryEventConfig) {
type MemoryEvent struct { type MemoryEvent struct {
*base *base
cfg *define.MemoryEventConfig cfg *define.MemoryEventConfig
messageChannel map[int]chan *define.EventData messageChannel chan *define.EventData
} }
// SendEvent 发送事件 // SendEvent 发送事件
@ -56,11 +55,10 @@ func (m *MemoryEvent) SendEvent(ctx context.Context, eventData *define.EventData
if nil == eventData { if nil == eventData {
return nil, errors.New("event data is nil") return nil, errors.New("event data is nil")
} }
partition := int(wrapper.String(eventData.Key).HashNumber().Value % uint64(m.cfg.PartitionNum)) m.messageChannel <- eventData
m.messageChannel[partition] <- eventData
return &define.SendResult{ return &define.SendResult{
Data: eventData, Data: eventData,
Partition: partition, Partition: 0,
Topic: "", Topic: "",
IsSuccess: true, IsSuccess: true,
FailReason: "success", FailReason: "success",
@ -109,8 +107,7 @@ func (m *MemoryEvent) SendEventAsync(ctx context.Context, eventData *define.Even
// //
// Date : 12:41 2024/7/17 // Date : 12:41 2024/7/17
func (m *MemoryEvent) GetConsumeEventChan() (<-chan *define.EventData, error) { func (m *MemoryEvent) GetConsumeEventChan() (<-chan *define.EventData, error) {
//TODO implement me return m.messageChannel, nil
panic("implement me")
} }
func (m *MemoryEvent) ConsumeEvent(handler abstract.EventHandler, successCallback abstract.ConsumeSuccessCallback, failureCallback abstract.ConsumeFailCallbackHandler) error { func (m *MemoryEvent) ConsumeEvent(handler abstract.EventHandler, successCallback abstract.ConsumeSuccessCallback, failureCallback abstract.ConsumeFailCallbackHandler) error {
@ -136,9 +133,6 @@ func (m *MemoryEvent) Init(cfg *define.MemoryEventConfig) {
if nil == cfg { if nil == cfg {
cfg = &define.MemoryEventConfig{} cfg = &define.MemoryEventConfig{}
} }
if cfg.PartitionNum <= 0 {
cfg.PartitionNum = define.DefaultMemoryPartitionNum
}
if cfg.MessageBufferSize <= 0 { if cfg.MessageBufferSize <= 0 {
cfg.MessageBufferSize = define.DefaultMemoryChannelSize cfg.MessageBufferSize = define.DefaultMemoryChannelSize
} }
@ -147,8 +141,5 @@ func (m *MemoryEvent) Init(cfg *define.MemoryEventConfig) {
} }
m.cfg = cfg m.cfg = cfg
// 初始化内存 channel // 初始化内存 channel
m.messageChannel = make(map[int]chan *define.EventData) m.messageChannel = make(chan *define.EventData, cfg.MessageBufferSize)
for num := 0; num < cfg.PartitionNum; num++ {
m.messageChannel[num] = make(chan *define.EventData, cfg.MessageBufferSize)
}
} }