blob: adedd1c22ec21f6e8e754381e944aba7d2418bfd (
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
|
package client
import (
"errors"
"sync"
"github.com/karyontech/karyon-go/jsonrpc/message"
)
var (
requestChannelNotFoundErr = errors.New("Request channel not found")
)
// messageDispatcher Is a structure that holds a map of request IDs and
// channels, and it is protected by mutex
type messageDispatcher struct {
sync.Mutex
chans map[message.RequestID]chan<- message.Response
}
// newMessageDispatcher Creates a new messageDispatcher
func newMessageDispatcher() *messageDispatcher {
chans := make(map[message.RequestID]chan<- message.Response)
return &messageDispatcher{
chans: chans,
}
}
// register Registers a new request channel with the given id. It returns a
// channel for receiving response.
func (c *messageDispatcher) register(key message.RequestID) <-chan message.Response {
c.Lock()
defer c.Unlock()
ch := make(chan message.Response)
c.chans[key] = ch
return ch
}
// dispatch Disptaches the response to the channel with the given request id
func (c *messageDispatcher) dispatch(key message.RequestID, res message.Response) error {
c.Lock()
defer c.Unlock()
if ch, ok := c.chans[key]; ok {
ch <- res
} else {
return requestChannelNotFoundErr
}
return nil
}
// unregister Unregisters the request with the provided id
func (c *messageDispatcher) unregister(key message.RequestID) {
c.Lock()
defer c.Unlock()
if ch, ok := c.chans[key]; ok {
close(ch)
delete(c.chans, key)
}
}
// clear Closes all the request channels and remove them from the map
func (c *messageDispatcher) clear() {
c.Lock()
defer c.Unlock()
for _, ch := range c.chans {
close(ch)
}
c.chans = nil
}
|
