From 38eccdf006c8374e89b7b7bb816b2d4ce4b2a220 Mon Sep 17 00:00:00 2001
From: liuxiaolong <736321739@qq.com>
Date: 星期二, 14 一月 2020 20:31:34 +0800
Subject: [PATCH] fix msg send
---
mangos.go | 28 ++++++++++++++++------------
1 files changed, 16 insertions(+), 12 deletions(-)
diff --git a/mangos.go b/mangos.go
index 169c2c4..47a9b0d 100644
--- a/mangos.go
+++ b/mangos.go
@@ -12,6 +12,7 @@
heartBeatUrl string
pubCh chan Message //publish msg chan
aliveNodes gopherdiscovery.StringSet
+ clients map[string][]string
recvCh chan Message //recv msg chan
}
@@ -36,6 +37,7 @@
heartBeatUrl: heartBeatUrl,
aliveNodes: gopherdiscovery.NewStringSet(),
pubCh: make(chan Message, 50),
+ clients: make(map[string][]string),
}
var msgCache = make(map[string]Message)
//clientMsgCh := make(map[string]chan Message)
@@ -55,15 +57,18 @@
discoveryServer.PublishMsg(string(sendB))
}
default:
- nodeIds := discoveryServer.AliveNodes().ToSlice()
- if len(nodeIds) >0 {
- //for _,nodeId := range nodeIds {
- // if _,ok := clientMsgCh[nodeId]; !ok {
- // clientMsgCh[nodeId] = make(chan Message)
- // }
- //}
+ nodeIds := discoveryServer.AliveNodes()
- if cacheNodes.Cardinality() == 0 { //绗竴娆℃湁涓婄嚎鐨勮妭鐐�
+ //for _,nodeId := range nodeIds {
+ // if _,ok := clientMsgCh[nodeId]; !ok {
+ // clientMsgCh[nodeId] = make(chan Message)
+ // }
+ //}
+ removedNodes := cacheNodes.Difference(nodeIds)
+ addedNodes := nodeIds.Difference(cacheNodes)
+ if len(nodeIds.ToSlice()) >0 {
+ if removedNodes.Cardinality() >0 || addedNodes.Cardinality() >0 { //鑺傜偣鏈夊彉鍖栫殑鏃跺�欙紝涔熼渶瑕佸彂娑堟伅
+ fmt.Println("removedNodes:", removedNodes, "addedNodes:", addedNodes)
if len(msgCache) > 0 {
for _,cMsg := range msgCache {
sendB, _ := json.Marshal(cMsg)
@@ -71,16 +76,15 @@
}
}
}
- cacheNodes = discoveryServer.AliveNodes()
- } else {
+ cacheNodes = nodeIds
+ } else {//璁㈤槄鑰呭叏閮ㄩ樀浜�
+ cacheNodes = nodeIds
time.Sleep(10 * time.Millisecond)
}
}
}
}()
- go func() {
- }()
return pub,nil
}
--
Gitblit v1.8.0