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
package eventstream
import (
"sync"
"time"
)
type seenEntry struct {
at time.Time
source string
drops int
settled bool
ok bool
done chan struct{}
}
type seenOrder struct {
key string
at time.Time
}
type duplicate struct {
First string
Drops int
Dropped int64
}
type seenSet struct {
window time.Duration
limit int
mu sync.Mutex
entries map[string]*seenEntry
order []seenOrder
dropped int64
}
func newSeenSet(window time.Duration, limit int) *seenSet {
return &seenSet{window: window, limit: limit, entries: map[string]*seenEntry{}}
}
func (s *seenSet) evictFront() {
front := s.order[0]
s.order = s.order[1:]
if e, ok := s.entries[front.key]; ok && e.at.Equal(front.at) {
delete(s.entries, front.key)
}
}
func (s *seenSet) claim(key, source string, now time.Time) (*seenEntry, bool) {
s.mu.Lock()
defer s.mu.Unlock()
for len(s.order) > 0 && now.Sub(s.order[0].at) >= s.window {
s.evictFront()
}
if e, found := s.entries[key]; found && (!e.settled || now.Sub(e.at) < s.window) {
return e, false
}
e := &seenEntry{at: now, source: source, done: make(chan struct{})}
s.entries[key] = e
s.order = append(s.order, seenOrder{key: key, at: now})
for len(s.order) > s.limit {
s.evictFront()
}
return e, true
}
func (s *seenSet) finish(key string, e *seenEntry, ok bool) {
s.mu.Lock()
defer s.mu.Unlock()
e.settled, e.ok = true, ok
if !ok && s.entries[key] == e {
delete(s.entries, key)
}
close(e.done)
}
func (s *seenSet) drop(e *seenEntry) duplicate {
s.mu.Lock()
defer s.mu.Unlock()
e.drops++
s.dropped++
return duplicate{First: e.source, Drops: e.drops, Dropped: s.dropped}
}
