-
Notifications
You must be signed in to change notification settings - Fork 0
/
consumer.go
159 lines (132 loc) · 2.78 KB
/
consumer.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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
package e2e
import (
"time"
"github.com/g41797/sputnik"
"github.com/g41797/sputnik/sidecar"
)
const (
syslogConsumerName = "syslogconsumer"
syslogConsumerResponsibility = "syslogconsumer"
)
func syslogConsumerDescriptor() sputnik.BlockDescriptor {
return sputnik.BlockDescriptor{Name: syslogConsumerName, Responsibility: syslogConsumerResponsibility}
}
func init() {
sputnik.RegisterBlockFactory(syslogConsumerName, syslogConsumerBlockFactory)
}
func syslogConsumerBlockFactory() *sputnik.Block {
cons := new(consumer)
if mcf == nil {
return nil
}
mc := mcf()
if mc == nil {
return nil
}
cons.mc = mc
block := sputnik.NewBlock(
sputnik.WithInit(cons.init),
sputnik.WithRun(cons.run),
sputnik.WithFinish(cons.finish),
sputnik.WithOnConnect(cons.brokerConnected),
sputnik.WithOnDisconnect(cons.brokerDisconnected),
)
return block
}
type consumer struct {
connected bool
cfact sputnik.ConfFactory
mc sidecar.MessageConsumer
sender sputnik.BlockCommunicator
shared sputnik.ServerConnection
stop chan struct{}
done chan struct{}
conn chan sputnik.ServerConnection
dscn chan struct{}
}
// Init
func (cons *consumer) init(fact sputnik.ConfFactory) error {
cons.cfact = fact
cons.stop = make(chan struct{}, 1)
cons.done = make(chan struct{}, 1)
cons.conn = make(chan sputnik.ServerConnection, 1)
cons.dscn = make(chan struct{}, 1)
return nil
}
// Finish:
func (cons *consumer) finish(init bool) {
if init {
return
}
close(cons.stop) // Cancel Run
<-cons.done // Wait finish of Run
return
}
// OnServerConnect:
func (cons *consumer) brokerConnected(srvc sputnik.ServerConnection) {
cons.conn <- srvc
return
}
// OnServerDisconnect:
func (cons *consumer) brokerDisconnected() {
cons.dscn <- struct{}{}
return
}
// Run
func (cons *consumer) run(bc sputnik.BlockCommunicator) {
cons.sender, _ = bc.Communicator(syslogClientResponsibility)
defer close(cons.done)
defer cons.disconnect()
waitBroker:
for {
select {
case <-cons.stop:
return
case cons.shared = <-cons.conn:
break waitBroker
}
}
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
waitConsumer:
for {
select {
case <-cons.stop:
return
case <-cons.dscn:
return
case <-ticker.C:
if cons.mc.Connect(cons.cfact, cons.shared) == nil {
cons.mc.Consume(cons.sender)
break waitConsumer
}
}
}
cons.runLoop()
return
}
func (cons *consumer) runLoop() {
loop:
for {
select {
case <-cons.stop:
break loop
case <-cons.dscn:
break loop
}
}
return
}
func (cons *consumer) disconnect() {
if cons == nil {
return
}
if cons.mc == nil {
return
}
cons.mc.Disconnect()
}
func RegisterMessageConsumerFactory(fact func() sidecar.MessageConsumer) {
mcf = fact
}
var mcf func() sidecar.MessageConsumer