-
Notifications
You must be signed in to change notification settings - Fork 609
/
wait.go
70 lines (61 loc) · 1.1 KB
/
wait.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
package raft
import (
"fmt"
"sync"
)
type waitItem struct {
// channel to wait up the waiter
ch chan interface{}
// callback which is called synchronously when the wait is triggered
cb func()
}
type wait struct {
l sync.Mutex
m map[uint64]waitItem
}
func newWait() *wait {
return &wait{m: make(map[uint64]waitItem)}
}
func (w *wait) register(id uint64, cb func()) <-chan interface{} {
w.l.Lock()
defer w.l.Unlock()
_, ok := w.m[id]
if !ok {
ch := make(chan interface{}, 1)
w.m[id] = waitItem{ch: ch, cb: cb}
return ch
}
panic(fmt.Sprintf("duplicate id %x", id))
}
func (w *wait) trigger(id uint64, x interface{}) bool {
w.l.Lock()
waitItem, ok := w.m[id]
delete(w.m, id)
w.l.Unlock()
if ok {
if waitItem.cb != nil {
waitItem.cb()
}
waitItem.ch <- x
close(waitItem.ch)
return true
}
return false
}
func (w *wait) cancel(id uint64) {
w.l.Lock()
waitItem, ok := w.m[id]
delete(w.m, id)
w.l.Unlock()
if ok {
close(waitItem.ch)
}
}
func (w *wait) cancelAll() {
w.l.Lock()
defer w.l.Unlock()
for id, waitItem := range w.m {
delete(w.m, id)
close(waitItem.ch)
}
}