/
util.go
65 lines (53 loc) · 1.05 KB
/
util.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
package core
import (
"sync"
)
type topic struct {
m *sync.Mutex
subscribers []chan<- string
}
func newTopic() *topic {
return &topic{
m: &sync.Mutex{},
subscribers: make([]chan<- string, 0, 10),
}
}
func (t *topic) Subscribe(subscription chan<- string) {
t.m.Lock()
defer t.m.Unlock()
t.subscribers = append(t.subscribers, subscription)
}
func (t *topic) Publish(msg string) {
t.m.Lock()
defer t.m.Unlock()
for _, s := range t.subscribers {
select {
case s <- msg:
default:
}
}
}
func (t *topic) Unsubscribe(unsubscribe chan<- string) {
t.m.Lock()
defer t.m.Unlock()
if i := find(t.subscribers, unsubscribe); i >= 0 {
// replace the unsubscribed with the 1st element and drop the 1st element
t.subscribers[i] = t.subscribers[0]
t.subscribers = t.subscribers[1:]
}
}
func find(subs []chan<- string, target chan<- string) int {
for i, s := range subs {
if s == target {
return i
}
}
return -1
}
func (t *topic) Close() {
t.m.Lock()
defer t.m.Unlock()
for _, s := range t.subscribers {
close(s)
}
}