forked from myzhan/boomer
-
Notifications
You must be signed in to change notification settings - Fork 0
/
client_czmq.go
130 lines (110 loc) · 2.74 KB
/
client_czmq.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
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
// +build goczmq
package boomer
import (
"fmt"
"log"
"github.com/zeromq/goczmq"
)
type czmqSocketClient struct {
masterHost string
masterPort int
identity string
dealerSocket *goczmq.Sock
fromMaster chan message
toMaster chan message
disconnectedFromMaster chan bool
shutdownChan chan bool
}
func newClient(masterHost string, masterPort int, identity string) (client *czmqSocketClient) {
log.Println("Boomer is built with goczmq support.")
client = &czmqSocketClient{
masterHost: masterHost,
masterPort: masterPort,
identity: identity,
fromMaster: make(chan message, 100),
toMaster: make(chan message, 100),
disconnectedFromMaster: make(chan bool),
shutdownChan: make(chan bool),
}
return client
}
func (c *czmqSocketClient) connect() (err error) {
addr := fmt.Sprintf("tcp://%s:%d", c.masterHost, c.masterPort)
dealer := goczmq.NewSock(goczmq.Dealer)
dealer.SetOption(goczmq.SockSetIdentity(c.identity))
err = dealer.Connect(addr)
if err != nil {
return err
}
c.dealerSocket = dealer
log.Printf("Boomer is connected to master(%s) press Ctrl+c to quit.\n", addr)
go c.recv()
go c.send()
return nil
}
func (c *czmqSocketClient) close() {
close(c.shutdownChan)
if c.dealerSocket != nil {
c.dealerSocket.Destroy()
}
}
func (c *czmqSocketClient) recvChannel() chan message {
return c.fromMaster
}
func (c *czmqSocketClient) recv() {
for {
select {
case <-c.shutdownChan:
return
default:
msg, _, err := c.dealerSocket.RecvFrame()
if err != nil {
log.Printf("Error reading: %v\n", err)
continue
}
decodedMsg, err := newGenericMessageFromBytes(msg)
if err != nil {
log.Printf("Msgpack decode fail: %v\n", err)
continue
}
if decodedMsg.NodeID != c.identity {
log.Printf("Recv a %s message for node(%s), not for me(%s), dropped.\n", decodedMsg.Type, decodedMsg.NodeID, c.identity)
continue
}
c.fromMaster <- decodedMsg
}
}
}
func (c *czmqSocketClient) sendChannel() chan message {
return c.toMaster
}
func (c *czmqSocketClient) send() {
for {
select {
case <-c.shutdownChan:
return
case msg := <-c.toMaster:
c.sendMessage(msg)
m, ok := msg.(*genericMessage)
if ok {
if m.Type == "quit" {
c.disconnectedFromMaster <- true
}
}
}
}
}
func (c *czmqSocketClient) sendMessage(msg message) {
serializedMessage, err := msg.serialize()
if err != nil {
log.Printf("Msgpack encode fail: %v\n", err)
return
}
err = c.dealerSocket.SendFrame(serializedMessage, goczmq.FlagNone)
if err != nil {
log.Printf("Error sending: %v\n", err)
}
}
func (c *czmqSocketClient) disconnectedChannel() chan bool {
return c.disconnectedFromMaster
}