/
client.go
143 lines (118 loc) · 3.23 KB
/
client.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
131
132
133
134
135
136
137
138
139
140
141
142
143
package CoapPubsub
import (
"errors"
"log"
"time"
"github.com/dustin/go-coap"
)
type subConnection struct {
channel chan string
clientCon *coap.Conn
}
type CoapPubsubClient struct {
msgIndex uint16
serAddr string
subList map[string]subConnection
}
// Create a pubsub client for CoAP protocol
// It will connect to server and make sure it alive and start heart beat
// To keep udp port open, we will send heart beat event to server every minutes
func NewCoapPubsubClient(servAddr string) *CoapPubsubClient {
c := new(CoapPubsubClient)
c.subList = make(map[string]subConnection, 0)
c.serAddr = servAddr
//TODO: connection check if any error
//Start heart beat
c.msgIndex = GetIPv4Int16() + GetLocalRandomInt()
log.Println("Init msgID=", c.msgIndex)
go c.heartBeat()
return c
}
//AddSub adds Subscription on topic and return a channel for user to wait data
func (c *CoapPubsubClient) AddSub(topic string) (chan string, error) {
if val, exist := c.subList[topic]; exist {
//if topic already exist in sub, return and not send to server
return val.channel, nil
}
conn, err := c.sendPubsubReq("ADDSUB", topic)
if err != nil {
return nil, err
}
subChan := make(chan string)
go c.waitSubResponse(conn, subChan, topic)
//Add client connection into member variable for heart beat
clientConn := subConnection{channel: subChan, clientCon: conn}
c.subList[topic] = clientConn
return subChan, nil
}
//RemoveSub removes Subscribetion on topic
func (c *CoapPubsubClient) RemoveSub(topic string) error {
if _, exist := c.subList[topic]; !exist {
//if topic not in sub list, return and not send to server
return nil
}
_, err := c.sendPubsubReq("REMSUB", topic)
return err
}
func (c *CoapPubsubClient) sendPubsubReq(cmd string, topic string) (*coap.Conn, error) {
Req := coap.Message{
Type: coap.Confirmable,
Code: coap.GET,
MessageID: c.getMsgID(),
Payload: []byte(""),
}
Req.SetOption(coap.ETag, cmd)
Req.SetPathString(topic)
conn, err := coap.Dial("udp", c.serAddr)
if err != nil {
log.Printf(cmd, ">>Error dialing: %v \n", err)
return nil, errors.New("Dial failed")
}
conn.Send(Req)
return conn, err
}
func (c *CoapPubsubClient) waitSubResponse(conn *coap.Conn, ch chan string, topic string) {
var rv *coap.Message
var err error
var keepLoop bool
keepLoop = true
for keepLoop {
if rv != nil {
if err != nil {
log.Fatalf("Error receiving: %v", err)
}
log.Printf("Got %s", rv.Payload)
}
rv, err = conn.Receive()
if err == nil {
ch <- string(rv.Payload)
}
time.Sleep(time.Second)
if _, exist := c.subList[topic]; !exist {
//sub topic already remove, leave loop
log.Println("Loop topic:", topic, " already remove leave loop")
keepLoop = false
}
}
}
func (c *CoapPubsubClient) getMsgID() uint16 {
c.msgIndex = c.msgIndex + 1
return c.msgIndex
}
func (c *CoapPubsubClient) heartBeat() {
log.Println("Starting heart beat loop call")
hbReq := coap.Message{
Type: coap.Confirmable,
Code: coap.GET,
MessageID: c.getMsgID(),
Payload: []byte("Heart beat msg."),
}
hbReq.SetOption(coap.ETag, "HB")
for {
for k, conn := range c.subList {
conn.clientCon.Send(hbReq)
log.Println("Send the heart beat in topic ", k)
}
time.Sleep(time.Minute)
}
}