/
state.go
108 lines (89 loc) · 2.31 KB
/
state.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
package sdk
import (
"errors"
"github.com/Chendemo12/micromq/src/proto"
"sync"
"time"
)
//goland:noinspection GoUnusedGlobalVariable
var NewQueue = proto.NewQueue
var (
framePool = proto.NewFramePool()
hmPool = NewHCPMPool()
emPool = proto.NewCPMPool()
)
//goland:noinspection GoUnusedGlobalVariable
var (
ErrTopicEmpty = errors.New("topic is empty")
ErrConsumerHandlerIsNil = errors.New("consumer messageHandler is nil")
ErrConsumerUnregistered = errors.New("consumer unregistered")
ErrConsumerUnconnected = errors.New("consumer unconnected")
ErrProducerUnregistered = errors.New("producer unregistered")
ErrProducerUnconnected = errors.New("producer unconnected")
ErrTokenIncorrect = errors.New("token incorrect")
)
const (
DefaultProducerSendInterval = 500 * time.Millisecond
)
//goland:noinspection GoUnusedGlobalVariable
const (
AllConfirm = proto.AllConfirm
NoConfirm = proto.NoConfirm
LeaderConfirm = proto.LeaderConfirm
)
type Queue = proto.Queue
type HCPMessagePool struct {
cpool *sync.Pool
ppool *sync.Pool
cCounter *proto.Counter
pCounter *proto.Counter
}
// CMHistoryNum ConsumerMessage 历史数量
func (m *HCPMessagePool) CMHistoryNum() uint64 { return m.cCounter.Value() }
// PMHistoryNum ProducerMessage 历史数量
func (m *HCPMessagePool) PMHistoryNum() uint64 { return m.pCounter.Value() }
func (m *HCPMessagePool) GetCM() *ConsumerMessage {
v := m.cpool.Get().(*ConsumerMessage)
v.Reset()
v.counter = m.cCounter.ValueBeforeIncrement()
return v
}
func (m *HCPMessagePool) PutCM(v *ConsumerMessage) {
cv := m.cCounter.Value()
if cv == 0 || v.counter <= cv {
v.Reset()
m.cpool.Put(v)
}
}
func (m *HCPMessagePool) GetPM() *ProducerMessage {
v := m.ppool.Get().(*ProducerMessage)
v.Reset()
v.counter = m.pCounter.ValueBeforeIncrement()
return v
}
func (m *HCPMessagePool) PutPM(v *ProducerMessage) {
cv := m.pCounter.Value()
if cv == 0 || v.counter <= cv {
v.Reset()
m.ppool.Put(v)
}
}
func NewHCPMPool() *HCPMessagePool {
p := &HCPMessagePool{
cpool: &sync.Pool{},
ppool: &sync.Pool{},
cCounter: proto.NewCounter(),
pCounter: proto.NewCounter(),
}
p.cpool.New = func() any {
cm := &ConsumerMessage{}
cm.Reset()
return cm
}
p.ppool.New = func() any {
pm := &ProducerMessage{}
pm.Reset()
return pm
}
return p
}