blob: d6d470a5f57dc8fddf3679ff23806a0268ba4735 (
plain)
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
|
package engine
import (
"sync"
"time"
)
const TickDuration = 600 * time.Millisecond
type Callback func() bool
type subscriber struct {
id uint64
interval int
callback Callback
ticks int
}
type Engine struct {
mu sync.Mutex
subscribers map[uint64]*subscriber
nextID uint64
ticker *time.Ticker
running bool
stopCh chan struct{}
}
func New() *Engine {
return &Engine{
subscribers: make(map[uint64]*subscriber),
}
}
func (e *Engine) Subscribe(interval int, cb Callback) uint64 {
e.mu.Lock()
defer e.mu.Unlock()
e.nextID++
e.subscribers[e.nextID] = &subscriber{
id: e.nextID,
interval: interval,
callback: cb,
}
return e.nextID
}
func (e *Engine) Unsubscribe(id uint64) {
e.mu.Lock()
defer e.mu.Unlock()
delete(e.subscribers, id)
}
func (e *Engine) Start() {
e.mu.Lock()
defer e.mu.Unlock()
if e.running {
return
}
e.running = true
e.ticker = time.NewTicker(TickDuration)
e.stopCh = make(chan struct{})
go func() {
for {
select {
case <-e.ticker.C:
e.processTick()
case <-e.stopCh:
return
}
}
}()
}
func (e *Engine) Stop() {
e.mu.Lock()
defer e.mu.Unlock()
if e.ticker != nil {
e.ticker.Stop()
}
e.running = false
if e.stopCh != nil {
close(e.stopCh)
}
}
func (e *Engine) processTick() {
e.mu.Lock()
snapshot := make(map[uint64]*subscriber, len(e.subscribers))
for id, sub := range e.subscribers {
snapshot[id] = sub
}
e.mu.Unlock()
for id, sub := range snapshot {
sub.ticks++
if sub.ticks >= sub.interval {
sub.ticks = 0
if !sub.callback() {
e.mu.Lock()
delete(e.subscribers, id)
e.mu.Unlock()
}
}
}
}
|
