/
bus.go
170 lines (147 loc) · 3.91 KB
/
bus.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
160
161
162
163
164
165
166
167
168
169
170
package events
import (
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
)
// EventBus manages the state of and transmits messages to all its Subscribers
type EventBus struct {
registry map[*Subscriber]bool
lock *sync.RWMutex
reload bool
done sync.WaitGroup
// circular buffer of events
head int
tail int
buf []Event
}
func (bus *EventBus) enqueue(event Event) {
bus.buf[bus.mod(bus.head+1)] = event
old := bus.head
bus.head = (bus.head + 1) % len(bus.buf)
if old != -1 && bus.head == bus.tail {
bus.tail = bus.mod(bus.tail + 1)
}
}
// DebugEvents ...
func (bus *EventBus) DebugEvents() []Event {
time.Sleep(100 * time.Millisecond)
events := []Event{}
for {
if bus.head == -1 {
break
}
event := bus.buf[bus.mod(bus.tail)]
if bus.tail == bus.head {
bus.head = -1
bus.tail = 0
} else {
bus.tail = bus.mod(bus.tail + 1)
}
if event == NonEvent {
break
}
events = append(events, event)
}
return events
}
func (bus *EventBus) mod(p int) int {
return p % len(bus.buf)
}
var collector *prometheus.CounterVec
func init() {
collector = prometheus.NewCounterVec(prometheus.CounterOpts{
Name: "containerpilot_events",
Help: "count of ContainerPilot events, partitioned by type and source",
}, []string{"code", "source"})
prometheus.MustRegister(collector)
}
// NewEventBus initializes an EventBus. We need this rather than a struct
// literal so that we know our channels are non-nil (which block sends).
func NewEventBus() *EventBus {
lock := &sync.RWMutex{}
reg := make(map[*Subscriber]bool)
buf := make([]Event, 10)
for i := range buf {
buf[i] = Event{}
}
return &EventBus{
registry: reg,
lock: lock,
buf: buf,
head: -1,
tail: 0,
reload: false,
}
}
// Register the Publisher for all Events
func (bus *EventBus) Register(publisher EventPublisher) {
bus.lock.Lock()
defer bus.lock.Unlock()
bus.done.Add(1)
}
// Unregister the Publisher from all Events
func (bus *EventBus) Unregister(publisher EventPublisher) {
bus.lock.Lock()
defer bus.lock.Unlock()
bus.done.Done()
}
// Subscribe the Subscriber for all Events
func (bus *EventBus) Subscribe(subscriber EventSubscriber) {
bus.lock.Lock()
defer bus.lock.Unlock()
sub := subscriber.(*Subscriber)
bus.registry[sub] = true
bus.done.Add(1)
}
// Unsubscribe the Subscriber from all Events
func (bus *EventBus) Unsubscribe(subscriber EventSubscriber) {
bus.lock.Lock()
defer bus.lock.Unlock()
sub := subscriber.(*Subscriber)
if _, ok := bus.registry[sub]; ok {
delete(bus.registry, sub)
}
bus.done.Done()
}
// Publish an Event to all Subscribers
func (bus *EventBus) Publish(event Event) {
bus.lock.Lock()
defer bus.lock.Unlock()
log.Debugf("event: %v", event)
if event.Code.String() != "Metric" {
collector.WithLabelValues(event.Code.String(), event.Source).Inc()
}
for subscriber := range bus.registry {
// sending to an unsubscribed Subscriber shouldn't be a runtime
// error, so this is in intentionally allowed to panic here
subscriber.Receive(event)
}
bus.enqueue(event)
}
// PublishSignal publishes a signal event through the EventBus to any Jobs that
// are subscribed to trigger on them.
func (bus *EventBus) PublishSignal(sig string) {
bus.Publish(Event{Code: Signal, Source: sig})
}
// SetReloadFlag sets the flag that Wait will use to signal to the main
// App that we want to restart rather than be shut down
func (bus *EventBus) SetReloadFlag() {
bus.lock.Lock()
defer bus.lock.Unlock()
bus.reload = true
}
// Shutdown asks all Subscribers to halt by sending the GlobalShutdown
// message. Subscribers are responsible for handling this message.
func (bus *EventBus) Shutdown() {
bus.Publish(GlobalShutdown)
}
// Wait blocks until the EventBus registry is unpopulated. Returns true
// if the "reload" flag was set.
func (bus *EventBus) Wait() bool {
bus.done.Wait()
bus.lock.RLock()
defer bus.lock.RUnlock()
return bus.reload
}