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
|
package serf
type latestUserEvents struct {
LTime LamportTime
Events []Event
}
type userEventCoalescer struct {
// Maps an event name into the latest versions
events map[string]*latestUserEvents
}
func (c *userEventCoalescer) Handle(e Event) bool {
// Only handle EventUser messages
if e.EventType() != EventUser {
return false
}
// Check if coalescing is enabled
user := e.(UserEvent)
return user.Coalesce
}
func (c *userEventCoalescer) Coalesce(e Event) {
user := e.(UserEvent)
latest, ok := c.events[user.Name]
// Create a new entry if there are none, or
// if this message has the newest LTime
if !ok || latest.LTime < user.LTime {
latest = &latestUserEvents{
LTime: user.LTime,
Events: []Event{e},
}
c.events[user.Name] = latest
return
}
// If the the same age, save it
if latest.LTime == user.LTime {
latest.Events = append(latest.Events, e)
}
}
func (c *userEventCoalescer) Flush(outChan chan<- Event) {
for _, latest := range c.events {
for _, e := range latest.Events {
outChan <- e
}
}
c.events = make(map[string]*latestUserEvents)
}
|
