fix msg send

liuxiaolong

2020-01-14 38eccdf006c8374e89b7b7bb816b2d4ce4b2a220

fix msg send

 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)
@@ -62,9 +64,11 @@
                //        clientMsgCh[nodeId] = make(chan Message)
                //    }
                //}
                removedNodes := cacheNodes.Difference(nodeIds)
                addedNodes := nodeIds.Difference(cacheNodes)
                if len(nodeIds.ToSlice()) >0 {
                    if len(nodeIds.Difference(cacheNodes).ToSlice()) > 0 { //节点有变化的时候，也需要发消息
                        fmt.Println("aliveNodes:",nodeIds)
                    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)
@@ -72,9 +76,9 @@
                            }
                        }
                    }
                    cacheNodes = discoveryServer.AliveNodes()
                    cacheNodes = nodeIds
                } else {//订阅者全部阵亡
                    cacheNodes = discoveryServer.AliveNodes()
                    cacheNodes = nodeIds
                    time.Sleep(10 * time.Millisecond)
                }
            }