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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
package pool
import (
"errors"
"fmt"
"github.com/gorilla/websocket"
"net/http"
"pool/pool/util/grpool"
"sync"
"time"
)
const (
// Time allowed to read the next pong message from the peer.
pongWait = 60 * time.Second
// Send pings to peer with this period. Must be less than pongWait.
pingPeriod = (pongWait * 9) / 10
// Time allowed to write a message to the peer.
writeWait = 10 * time.Second
// Maximum message size allowed from peer.
maxMessageSize = 1024 * 1024 * 20
//
closeWait = 60 * 30 * time.Second
// time for auto end
endDate = 60 * 10 * time.Second
)
var cliKey = "clients"
var convKey = "conversation"
var upgrader = websocket.Upgrader{
//ReadBufferSize: 1024 * 1024,
//WriteBufferSize: 1024 * 1024,
// 默认允许WebSocket请求跨域,权限控制可以由业务层自己负责,灵活度更高
CheckOrigin: func(r *http.Request) bool {
return true
},
}
//连接参数结构体
type Config struct {
Id string //标识连接的名称
Type string //连接类型或path
Channel []string //连接注册频道类型方便广播等操作。做为一个数组存储。因为一个连接可以属多个频道
Goroutine int //每个连接开启的go程数里 默认为10
}
type RuntimeInfo struct {
Id string //标识连接的名称
Type string //连接类型或path
Ip string
Channel []string //连接注册频道类型方便广播等操作。做为一个数组存储。因为一个连接可以属多个频道
OpenTime time.Time //连接打开时间
LastReceiveTime time.Time //最后一次接收到数据的时间
LastSendTime time.Time //最后一次发送数据的时间
}
type SendMsg struct {
ConversationId int `json:"conversationId"`
SendTime string `json:"sendTime"` //消息发送时间
FromClientId string `json:"fromClientId"` //指令消息的来源。发送者的连接ID
ToClientId string `json:"toClientId"` //指令消息的接收者。发送给对应的客户端连接ID
CmdData []byte `json:"cmdData"` //对应指令的CmdData1的protobuf的message
Status int `json:"status"` //消息发送响应状态
Priority int `json:"priority"` //用于处理指令队列的优先级的权重值
BusinessId string `json:"businessId"` //客户端标识消息的id,主要区分相同cmd的不同消息,方便收到回复分发处理等效果,考虑长度问题定义成string
GroupId string `json:"groupId"` //服务端发送消息的ID,主要区分相同cmd的不同消息,方便服务端收到回复分发处理等效果,考虑长度问题定义成string
Channel []string `json:"channel"` //指定需要广播的频道,可以指定一个或多个频道
ProcedureType int `json:"procedureType"` //会话状态 1 开始导诊 2 等待连接 3 离线 4 双方建立连接 5 结束会话 6 对话 7 已读 8 超时关闭
MsgType int `json:"msgType"` // msg类型 procedureType=6时填写 1 文本消息 2 图片消息
Msg string `json:"msg"` //一般用于json数据传递,或消息发送响应内容
Desc string `json:"desc"` //消息介绍内容,或其它数据
}
// Client is a middleman between the websocket connection and the hub.
type Client struct {
hub *hub
// The websocket connection.
conn *websocket.Conn
types string //连接类型或path
openTime time.Time //连接打开时间
CloseTime time.Time //连接断开的时间
lastReceiveTime time.Time //最后一次接收到数据的时间
lastSendTime time.Time //最后一次发送数据的时间
Id string //标识连接的名称
mux *sync.Mutex
IsClose chan bool //连接的状态。true为关闭
channel []string //连接注册频道类型方便广播等操作。做为一个数组存储。因为一个连接可以属多个频道
// Buffered channel of outbound messages.
grpool *grpool.Pool
sendCh chan *SendMsg //发送消息的缓冲管首
recvCh chan *SendMsg //接收消息的缓冲管首
recvPing chan int //收到ping的存储管道,方便回复pong处理
sendPing chan int //发送ping的存储管道,方便收到pong处理下次发ping
ticker *time.Ticker //定时发送ping的定时器
onError func(error)
onOpen func() //连接成功的回调
onPing func() //收到ping
onPong func() //收到pong
onMessage func(*SendMsg)
onClose func()
pingPeriodTicker *time.Timer
closeTicker *time.Timer
//ToSendData []ToSendData //连接方列表
}
//type ToSendData struct {
// toSendTime string //开启会话时间
// toSendId string //连接id
//}
type Conversation struct {
ConversationId int `json:"conversationId"`
Promoter string `json:"promoter"`
Participant string `json:"participant"`
StartTime string `json:"startTime"`
StartReceiveDate string `json:"startReceiveDate"`
Status int `json:"status"`
}
// readPump pumps messages from the websocket connection to the hub.
//
// The application runs readPump in a per-connection goroutine. The application
// ensures that there is at most one reader on a connection by executing all
// reads from this goroutine.
func (c *Client) readPump() {
defer func() {
fmt.Println("连接己关闭或者断开,正在清理对像")
c.conn.Close()
//触发连接关闭的事件回调
c.onClose() //先执行完关闭回调,再请空所有的回调
c.OnError(nil)
c.OnOpen(nil)
c.OnMessage(nil)
c.OnClose(nil)
c.OnPong(nil)
c.OnPing(nil)
close(c.recvPing)
close(c.sendPing)
c.grpool.Close()
c.hub.RemoveClient(c)
DelClient(c.Id)
dump()
}()
Loop:
for {
select {
case <-c.IsClose:
return
default:
_, message, err := c.conn.ReadMessage()
//if string(message) == "" {
// return
//}
if err != nil {
if websocket.IsUnexpectedCloseError(err,
websocket.CloseAbnormalClosure,
websocket.CloseGoingAway,
websocket.CloseProtocolError,
websocket.CloseUnsupportedData,
websocket.CloseNoStatusReceived,
websocket.CloseAbnormalClosure,
websocket.CloseInvalidFramePayloadData,
websocket.ClosePolicyViolation,
websocket.CloseMessageTooBig,
websocket.CloseMandatoryExtension,
websocket.CloseInternalServerErr,
websocket.CloseServiceRestart,
websocket.CloseTryAgainLater,
websocket.CloseTLSHandshake) {
fmt.Println("err:", err)
c.onError(errors.New("连接ID:" + c.Id + "ReadMessage Is Unexpected Close Error:" + err.Error()))
//c.closeChan<-true;
goto Loop1
}
c.onError(errors.New("连接ID:" + c.Id + "ReadMessage other error:" + err.Error()))
//c.closeChan<-true;
goto Loop1
}
c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.pingPeriodTicker.Reset(pingPeriod)
c.lastReceiveTime = time.Now()
//msg, err := unMarshal(message)
msg := &SendMsg{}
UnserislizeJson(message, msg)
if err != nil {
c.onError(errors.New("接收数据解析失败!!连接ID:" + c.Id + "原因:" + err.Error()))
break Loop
}
if msg.ProcedureType == 9 {
c.setPing()
}
if msg.ProcedureType == 10 {
c.setPong()
}
SaveMsg(msg)
//rabbitMQ.PublishSimple(string(message))
c.readMessage(msg)
}
}
Loop1:
c.close()
}
// 读取消息写管道缓冲区
func (c *Client) readMessage(msg *SendMsg) {
timeout := time.NewTimer(time.Millisecond * 800)
defer timeout.Stop()
select {
case <-c.IsClose:
c.onError(errors.New("readMessage连接" + c.Id + ",连接己在关闭,不进行消息接收"))
return
case c.recvCh <- msg:
return
case <-timeout.C:
c.onError(errors.New("recvCh 消息管道blocked,写入消息超时,管道长度:" + string(len(c.recvCh))))
return
}
}
// 单个连接接收消息
func (c *Client) recvMessage() {
defer func() {
dump()
}()
loop:
for {
select {
case <-c.IsClose:
return
case data, ok := <-c.recvCh:
if !ok {
break loop
}
//ToClientId与Channel不能同时存在!!!注意!!!!
//if message.ToClientId != "" {
// Send(message)
//}
//ToClientId与Channel不能同时存在!!!注意!!!!
//if message.Channel != "" {
// Broadcast(message)
//}
//收到消息触发回调
//c.onMessage(data)
c.grpool.Add(func() {
c.onMessage(data)
})
}
}
}
// writePump pumps messages from the hub to the websocket connection.
//
// A goroutine running writePump is started for each connection. The
// application ensures that there is at most one writer to a connection by
// executing all writes from this goroutine.
func (c *Client) writePump() {
defer func() {
dump()
}()
Loop:
for {
select {
case <-c.IsClose:
return
case d, ok := <-c.sendCh:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if !ok {
// The hub closed the channel.
//说明管道己经关闭
c.conn.WriteMessage(websocket.CloseMessage, []byte{})
//glog.Error("连接ID:"+c.Id,"wsServer发送消息失败,一般是连接channel已经被关闭:(此处服务端会断开连接,使客户端能够感知进行重连)")
goto Loop1
}
//message, err := marshal(d)
message := SerializeJson(d)
//if err != nil {
// c.onError(errors.New("接收数据ProtoBuf编码失败!!连接ID:" + c.Id + "原因:" + err.Error()))
// break Loop
//}
w, err := c.conn.NextWriter(websocket.TextMessage)
if err != nil {
goto Loop1
}
c.lastSendTime = time.Now()
_, err = w.Write(message)
if err != nil {
c.onError(errors.New("连接ID:" + c.Id + "写消息进写入IO错误!连接中断" + err.Error()))
goto Loop1
}
// Add queued chat messages to the current websocket message.
//n := len(c.sendCh)
//if n > 0 {
// for i := 0; i < n; i++ {
//
// _, err = w.Write(<-c.sendCh)
// if err != nil {
// c.onError(errors.New("连接ID:" + c.Id + "写上次连接未发送的消息消息进写入IO错误!连接中断" + err.Error()))
// return
// }
// }
//}
//关闭写入io对象
if err := w.Close(); err != nil {
c.onError(errors.New("连接ID:" + c.Id + "关闭写入IO对象出错,连接中断" + err.Error()))
goto Loop1
}
case p, ok := <-c.sendPing: //定时发送ping
if !ok {
break Loop
}
if p == 1 {
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.TextMessage, SerializeJson(&SendMsg{ProcedureType: 9})); err != nil {
fmt.Println(err)
c.onError(errors.New("连接ID:" + c.Id + "关闭写入IO对象出错,连接中断" + err.Error()))
goto Loop1
}
}
case p, ok := <-c.recvPing:
if !ok {
break Loop
}
if p == 1 {
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.TextMessage, SerializeJson(&SendMsg{ProcedureType: 10})); err != nil {
fmt.Println(err)
c.onError(errors.New("回复客户端PongMessage出现异常:" + err.Error()))
goto Loop1
}
}
}
}
Loop1:
c.close()
}
func (c *Client) setPong() {
c.conn.SetReadDeadline(time.Now().Add(pongWait))
fmt.Println("收到pong---", c.Id)
c.pingPeriodTicker.Reset(pingPeriod)
c.onPong()
}
func (c *Client) setPing() {
c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.pingPeriodTicker.Reset(pingPeriod)
fmt.Println("收到ping---", c.Id)
c.recvPing <- 1
//if err := c.conn.WriteMessage(websocket.PongMessage, nil); err != nil {
// c.onError(errors.New("回复客户端PongMessage出现异常:"+err.Error()))
//}
c.onPing()
}
func (c *Client) send(msg *SendMsg) {
timeout := time.NewTimer(time.Millisecond * 800)
defer timeout.Stop()
select {
case c.sendCh <- msg:
return
case <-timeout.C:
c.onError(errors.New("sendCh消息管道blocked,写入消息超时,管道长度:" + string(len(c.sendCh))))
return
}
//c.sendCh<-msg
}
//定时发送ping
func (c *Client) Tickers() {
for {
select {
case <-c.IsClose:
return
case <-c.pingPeriodTicker.C:
c.sendPing <- 1
}
}
}
func (c *Client) closeTickers() {
for {
select {
case <-c.IsClose:
return
case <-c.closeTicker.C:
return
}
}
}
func (c *Client) close() {
c.mux.Lock()
defer c.mux.Unlock()
select {
case <-c.IsClose:
return
default:
close(c.IsClose)
}
}