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
|
surveyors 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("newPub err:",err)
|
pub := &mangosPubSub{
|
url: publishUrl,
|
heartBeatUrl: heartBeatUrl,
|
surveyors: gopherdiscovery.NewStringSet(),
|
pubCh: make(chan Message, 50),
|
}
|
var msgCache = make(map[string]Message)
|
//clientMsgCh := make(map[string]chan Message)
|
go func() {
|
for {
|
select {
|
case msg := <-pub.pubCh:
|
msgCache[msg.Topic] = msg
|
//if len(clientMsgCh) > 0 {
|
// for _, ch := range clientMsgCh {
|
// ch <- msg
|
// }
|
//}
|
if pub.surveyors.Cardinality() >0 {
|
sendB, _ := json.Marshal(msg)
|
discoveryServer.PublishMsg(string(sendB))
|
}
|
default:
|
nodeIds := discoveryServer.AliveNodes()
|
|
//for _,nodeId := range nodeIds {
|
// if _,ok := clientMsgCh[nodeId]; !ok {
|
// clientMsgCh[nodeId] = make(chan Message)
|
// }
|
//}
|
removedNodes := pub.surveyors.Difference(nodeIds)
|
addedNodes := nodeIds.Difference(pub.surveyors)
|
if len(nodeIds.ToSlice()) >0 {
|
if addedNodes.Cardinality() >0 { //有新节点上线的时候，需要发一次消息，节点离线的时候不用管
|
fmt.Println("removedNodes:", removedNodes, "addedNodes:", addedNodes)
|
if len(msgCache) > 0 {
|
for _,cMsg := range msgCache {
|
sendB, _ := json.Marshal(cMsg)
|
discoveryServer.PublishMsg(string(sendB))
|
}
|
}
|
}
|
pub.surveyors = nodeIds
|
} else {//订阅者全部阵亡
|
pub.surveyors = nodeIds
|
time.Sleep(10 * time.Millisecond)
|
}
|
}
|
}
|
}()
|
|
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),
|
surveyors: gopherdiscovery.NewStringSet(),
|
}
|
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) Surveyor() []string {
|
return ps.surveyors.ToSlice()
|
}
|
|
func (ps *mangosPubSub) Publish(msg Message) {
|
ps.pubCh <- msg
|
}
|
|
func (ps *mangosPubSub) Recv() chan Message {
|
return ps.recvCh
|
}
|
