Skip to content

Commit 892647b

Browse files
syhlionsyhlion
committed
add ws reply message add field socket_id
1 parent bc6e274 commit 892647b

File tree

4 files changed

+41
-14
lines changed

4 files changed

+41
-14
lines changed

‎CHANGELOG.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,11 @@
44
### [Fix]
55

66

7+
## [v1.12.0]
8+
9+
- add websocket reply message add field socket_id
10+
11+
712
## [v1.11.0]
813

914
- add remote reponse switch

‎doc/protocal.md

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ reply scuess:
4848
```
4949
{
5050
"event":"gusher.querychannel_succeeded",
51+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
5152
"data":{
5253
"channels":["AA","BB"]
5354
}
@@ -58,6 +59,7 @@ reply error:
5859
```
5960
{
6061
"event":"gusher.querychannel_error",
62+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
6163
"data":{}
6264
}
6365
```
@@ -78,6 +80,7 @@ reply scuess:
7880
```
7981
{
8082
"event":"gusher.multi_subscribe_succeeded",
83+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
8184
"data":{
8285
"multi_channel":[]
8386
}
@@ -88,6 +91,7 @@ reply error:
8891
```
8992
{
9093
"event":"gusher.multi_subscribe_error",
94+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
9195
"data":{
9296
"multi_channel":[]
9397
}
@@ -109,6 +113,7 @@ reply scuess:
109113
```
110114
{
111115
"event":"gusher.subscribe_succeeded",
116+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
112117
"data":{
113118
"channel":""
114119
}
@@ -119,6 +124,7 @@ reply error:
119124
```
120125
{
121126
"event":"gusher.subscribe_error",
127+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
122128
"data":{
123129
"channel":""
124130
}
@@ -141,6 +147,7 @@ reply scuess:
141147
```
142148
{
143149
"event":"gusher.unsubscribe_succeeded",
150+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
144151
"data":{
145152
"channel":""
146153
}
@@ -151,6 +158,7 @@ reply error:
151158
```
152159
{
153160
"event":"gusher.unsubscribe_error",
161+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
154162
"data":{
155163
"channel":"",
156164
}
@@ -176,6 +184,7 @@ reply scuess:
176184
```
177185
{
178186
"event":"gusher.remote_succeeded",
187+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
179188
"data":{
180189
"remote":"test"
181190
"uid":"",
@@ -188,6 +197,7 @@ reply error:
188197
```
189198
{
190199
"event":"gusher.remote_error",
200+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
191201
"data":{
192202
"remote":"test",
193203
"uid":"",
@@ -225,6 +235,7 @@ reply :
225235
```
226236
{
227237
"event":"gusher.pong_succeeded",
238+
"socket_id":"cd19cdaa-44f1-11eb-80c2-784f43873ba3",
228239
"data":{
229240
//custom
230241
}

‎slave_api.go

Lines changed: 23 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,7 @@ func WtfConnect(sc SlaveConfig, pool *redis.Pool, jobPool *redis.Pool, rHub *red
121121
logger.WithFields(logrus.Fields{
122122
"data": string(data),
123123
}).Info("receive start")
124-
h, err := CommanRouter(data, jobPool, s.SocketId())
124+
h, err := CommanRouter(data, jobPool)
125125
if err != nil {
126126
logger.WithFields(logrus.Fields{
127127
"data": string(data),
@@ -140,7 +140,7 @@ func WtfConnect(sc SlaveConfig, pool *redis.Pool, jobPool *redis.Pool, rHub *red
140140
"data": string(data),
141141
"pdata": string(d),
142142
}).Info("receive to sub")
143-
res, err := h(appKey, s.GetAuth(), d, true)
143+
res, err := h(appKey, s.GetAuth(), d, s.SocketId(), true)
144144
if err != nil {
145145
logger.WithFields(logrus.Fields{
146146
"data": string(data),
@@ -208,7 +208,7 @@ func WsConnect(sc SlaveConfig, pool *redis.Pool, jobPool *redis.Pool, rHub *redi
208208
defer s.Close()
209209

210210
s.Listen(func(data []byte) (b []byte, err error) {
211-
h, err := CommanRouter(data, jobPool, s.SocketId())
211+
h, err := CommanRouter(data, jobPool)
212212
if err != nil {
213213
logger.WithError(err).Info("router error")
214214
return
@@ -222,7 +222,7 @@ func WsConnect(sc SlaveConfig, pool *redis.Pool, jobPool *redis.Pool, rHub *redi
222222
if err != nil {
223223
debug = false
224224
}
225-
res, err := h(appKey, s.GetAuth(), d, debug)
225+
res, err := h(appKey, s.GetAuth(), d, s.SocketId(), debug)
226226
if err != nil {
227227
logger.WithError(err).Info("handler error")
228228
return
@@ -244,7 +244,7 @@ func WsConnect(sc SlaveConfig, pool *redis.Pool, jobPool *redis.Pool, rHub *redi
244244
}
245245

246246
}
247-
func MultiSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error) {
247+
func MultiSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error) {
248248

249249
multiChannel := make([]string, 0)
250250
_, err = jsonparser.ArrayEach(data, func(v []byte, dataType jsonparser.ValueType, offset int, err error) {
@@ -286,6 +286,7 @@ func MultiSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, deb
286286
if len(subChannels) > 0 {
287287
msg.multiData = subChannels
288288
command.Event = MultiSubscribeReplySucceeded
289+
command.SocketId = socketId
289290
command.Data.Channel = subChannels
290291
reply, err = json.Marshal(command)
291292
if err != nil {
@@ -297,6 +298,7 @@ func MultiSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, deb
297298
//TODO 需重構 不讓他進入訂閱模式
298299
msg.cmdType = ""
299300
command.Event = MultiSubscribeReplyError
301+
command.SocketId = socketId
300302
command.Data.Channel = multiChannel
301303
reply, err = json.Marshal(command)
302304
if err != nil {
@@ -309,7 +311,7 @@ func MultiSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, deb
309311
return
310312
}
311313

312-
func SubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error) {
314+
func SubscribeCommand(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error) {
313315

314316
channel, err := jsonparser.GetString(data, "channel")
315317
if err != nil {
@@ -339,6 +341,7 @@ func SubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug bo
339341
var reply []byte
340342
if exist {
341343
msg.data = channel
344+
command.SocketId = socketId
342345
command.Event = SubscribeReplySucceeded
343346
command.Data.Channel = channel
344347
reply, err = json.Marshal(command)
@@ -350,6 +353,7 @@ func SubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug bo
350353

351354
//TODO 需重構 不讓他進入訂閱模式
352355
msg.cmdType = ""
356+
command.SocketId = socketId
353357
command.Event = SubscribeReplyError
354358
command.Data.Channel = channel
355359
reply, err = json.Marshal(command)
@@ -362,14 +366,15 @@ func SubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug bo
362366

363367
return
364368
}
365-
func QueryChannelCommand(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error) {
369+
func QueryChannelCommand(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error) {
366370
msg = &commandResponse{
367371
handler: DefaultSubHandler,
368372
cmdType: "QUERYCHANNEL",
369373
}
370374

371375
command := &QueryChannelResponse{}
372376
command.Event = QueryChannelReplySucceeded
377+
command.SocketId = socketId
373378
command.Data = struct {
374379
Channels []string `json:"channels"`
375380
}{
@@ -384,14 +389,15 @@ func QueryChannelCommand(appkey string, auth redisocket.Auth, data []byte, debug
384389
return
385390
}
386391

387-
func PingPongCommand(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error) {
392+
func PingPongCommand(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error) {
388393
msg = &commandResponse{
389394
handler: DefaultSubHandler,
390395
cmdType: "PING",
391396
}
392397

393398
command := &PongResponse{}
394399
command.Event = QueryChannelReplySucceeded
400+
command.SocketId = socketId
395401
command.Data = data
396402
command.Time = time.Now().Unix()
397403

@@ -402,8 +408,8 @@ func PingPongCommand(appkey string, auth redisocket.Auth, data []byte, debug boo
402408
msg.msg = reply
403409
return
404410
}
405-
func Remote(pool *redis.Pool, socketId string) func(string, redisocket.Auth, []byte, bool) (msg *commandResponse, err error) {
406-
return func(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error) {
411+
func Remote(pool *redis.Pool) func(string, redisocket.Auth, []byte, string, bool) (msg *commandResponse, err error) {
412+
return func(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error) {
407413

408414
remote, err := jsonparser.GetString(data, "remote")
409415
if err != nil {
@@ -429,6 +435,7 @@ func Remote(pool *redis.Pool, socketId string) func(string, redisocket.Auth, []b
429435
//沒有這個remote 返回錯誤訊息不斷線
430436
if !ok || !b {
431437
command.Event = RemoteReplyError
438+
command.SocketId = socketId
432439
reply, err = json.Marshal(command)
433440
if err != nil {
434441
return
@@ -451,6 +458,7 @@ func Remote(pool *redis.Pool, socketId string) func(string, redisocket.Auth, []b
451458
return
452459
}
453460
command.Event = RemoteReplySucceeded
461+
command.SocketId = socketId
454462
reply, err = json.Marshal(command)
455463
if err != nil {
456464
return
@@ -463,7 +471,7 @@ func Remote(pool *redis.Pool, socketId string) func(string, redisocket.Auth, []b
463471
}
464472

465473
}
466-
func UnSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error) {
474+
func UnSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error) {
467475
channel, err := jsonparser.GetString(data, "channel")
468476
if err != nil {
469477
return
@@ -493,6 +501,7 @@ func UnSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug
493501
if exist {
494502
msg.data = channel
495503
command.Event = UnSubscribeReplySucceeded
504+
command.SocketId = socketId
496505
command.Data.Channel = channel
497506
reply, err = json.Marshal(command)
498507
if err != nil {
@@ -505,6 +514,7 @@ func UnSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug
505514
//TODO 需重構 先不讓他進入訂閱模式
506515
msg.cmdType = ""
507516
command.Event = UnSubscribeReplyError
517+
command.SocketId = socketId
508518
command.Data.Channel = channel
509519
reply, err = json.Marshal(command)
510520
if err != nil {
@@ -515,15 +525,15 @@ func UnSubscribeCommand(appkey string, auth redisocket.Auth, data []byte, debug
515525
return
516526
}
517527

518-
func CommanRouter(data []byte, pool *redis.Pool, socketId string) (fn func(appkey string, auth redisocket.Auth, data []byte, debug bool) (msg *commandResponse, err error), err error) {
528+
func CommanRouter(data []byte, pool *redis.Pool) (fn func(appkey string, auth redisocket.Auth, data []byte, socketId string, debug bool) (msg *commandResponse, err error), err error) {
519529

520530
val, err := jsonparser.GetString(data, "event")
521531
if err != nil {
522532
return
523533
}
524534
switch val {
525535
case RemoteEvent:
526-
return Remote(pool, socketId), nil
536+
return Remote(pool), nil
527537
case QueryChannelEvent:
528538
return QueryChannelCommand, nil
529539
case SubscribeEvent:

‎spec.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,8 @@ type BatchData struct {
3535
}
3636

3737
type InternalCommand struct {
38-
Event string `json:"event"`
38+
Event string `json:"event"`
39+
SocketId string `json:"socket_id"`
3940
}
4041
type RemoteCommand struct {
4142
InternalCommand

0 commit comments

Comments
 (0)