fix

liuxiaolong

2020-01-13 fb64156e926cfd98d0b4891543bdb47151272486

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
package pubsub
 
import (
    "basic.com/valib/gopherdiscovery.git"
    "encoding/json"
    "fmt"
    "time"
)
 
type mangosPubSub struct {
    url string
    heartBeatUrl string
    pubCh chan Message  //publish msg chan
    aliveNodes gopherdiscovery.StringSet
 
    recvCh chan Message  //recv msg chan
}
 
 
func newPub(publishUrl string,heartBeatUrl string) (*mangosPubSub,error) {
    var discoveryServer *gopherdiscovery.DiscoveryServer
    var err error
    var (
        defaultOpts = gopherdiscovery.Options{
            SurveyTime:   3 * time.Second,
            //RecvDeadline: 3 * time.Second,
            PollTime:     5 * time.Second,
        }
    )
 
    discoveryServer, err = gopherdiscovery.Server(heartBeatUrl, publishUrl, defaultOpts)
 
    fmt.Println("err:",err)
    pub := &mangosPubSub{
        url: publishUrl,
        heartBeatUrl: heartBeatUrl,
        aliveNodes: gopherdiscovery.NewStringSet(),
        pubCh: make(chan Message, 50),
    }
    var msgCache = make(map[string]Message)
    //clientMsgCh := make(map[string]chan Message)
    cacheNodes := gopherdiscovery.NewStringSet()
    go func() {
        for {
            select {
            case msg := <-pub.pubCh:
                msgCache[msg.Topic] = msg
                //if len(clientMsgCh) > 0 {
                //    for _, ch := range clientMsgCh {
                //        ch <- msg
                //    }
                //}
                if cacheNodes.Cardinality() >0 {
                    sendB, _ := json.Marshal(msg)
                    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)
                    //    }
                    //}
 
                    if cacheNodes.Cardinality() == 0 { //第一次有上线的节点
                        if len(msgCache) > 0 {
                            for _,cMsg := range msgCache {
                                sendB, _ := json.Marshal(cMsg)
                                discoveryServer.PublishMsg(string(sendB))
                            }
                        }
                    }
                    cacheNodes = discoveryServer.AliveNodes()
                } else {
                    time.Sleep(10 * time.Millisecond)
                }
            }
        }
    }()
    go func() {
 
    }()
    return pub,nil
}
 
func newSub(subcribeUrl string,heartBeatUrl string, topics []string,procId string) (*mangosPubSub,error) {
    client, err := gopherdiscovery.ClientWithSub(heartBeatUrl, subcribeUrl, procId)
    if err !=nil {
        return nil,err
    }
    heartMsg := client.HeartBeatMsg()
    _= <-heartMsg
    fmt.Println("heat beat with server success")
    sub := &mangosPubSub{
        url:subcribeUrl,
        heartBeatUrl: heartBeatUrl,
        recvCh: make(chan Message,50),
    }
    go func() {
        peers, _ := client.Peers()
        for msg := range peers {
            //判断是否是想要的主题消息
            var recvMsg Message
            if err := json.Unmarshal(msg, &recvMsg);err ==nil {
                if matchTopic(recvMsg.Topic, topics) {
                    sub.recvCh <- recvMsg
                }
            }
        }
    }()
 
    return sub,nil
}
 
func matchTopic(topic string,subTopics []string) bool {
    if subTopics ==nil && len(subTopics) ==0 {
        return true
    }
    for _,t := range subTopics {
        if topic == t {
            return true
        }
    }
    return false
}
 
func (ps *mangosPubSub) Publish(msg Message) {
    ps.pubCh <- msg
}
 
func (ps *mangosPubSub) Recv() chan Message {
    return ps.recvCh
}