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
|
package client
import (
"fmt"
"sync"
)
// messageDispatcher Is a generic structure that holds a map of keys and
// channels, and it is protected by mutex
type messageDispatcher[K comparable, V any] struct {
sync.Mutex
chans map[K]chan<- V
bufferSize int
}
// newMessageDispatcher Creates a new messageDispatcher
func newMessageDispatcher[K comparable, V any](bufferSize int) messageDispatcher[K, V] {
chans := make(map[K]chan<- V)
return messageDispatcher[K, V]{
chans: chans,
bufferSize: bufferSize,
}
}
// register Registers a new channel with a given key. It returns the receiving channel.
func (c *messageDispatcher[K, V]) register(key K) <-chan V {
c.Lock()
defer c.Unlock()
ch := make(chan V, c.bufferSize)
c.chans[key] = ch
return ch
}
// length Returns the number of channels
func (c *messageDispatcher[K, V]) length() int {
c.Lock()
defer c.Unlock()
return len(c.chans)
}
// disptach Disptaches the msg to the channel with the given key
func (c *messageDispatcher[K, V]) disptach(key K, msg V) error {
c.Lock()
defer c.Unlock()
if ch, ok := c.chans[key]; ok {
ch <- msg
return nil
}
return fmt.Errorf("Channel not found")
}
// unregister Unregisters the channel with the provided key
func (c *messageDispatcher[K, V]) unregister(key K) {
c.Lock()
defer c.Unlock()
if ch, ok := c.chans[key]; ok {
close(ch)
delete(c.chans, key)
}
}
// clear Closes all the channels and remove them from the map
func (c *messageDispatcher[K, V]) clear() {
c.Lock()
defer c.Unlock()
for k, ch := range c.chans {
close(ch)
delete(c.chans, k)
}
}
|
