package serf
|
|
import (
|
"basic.com/valib/bhomeclient.git"
|
"basic.com/valib/c_bhomebus.git/proto/source/bhome_msg"
|
"basic.com/valib/logger.git"
|
"basic.com/valib/serf.git/serf"
|
"encoding/json"
|
"errors"
|
"github.com/golang/protobuf/proto"
|
"reflect"
|
"strconv"
|
"sync"
|
"time"
|
"vamicro/config"
|
"vamicro/extend/util"
|
"vamicro/system-service/bhome_msg_dev"
|
)
|
|
type RequestSerfTopicMsg struct {
|
Topic string `json:"topic"`
|
Msg []byte `json:"msg"`
|
targetNodeIds []string `json:"targetNodeIds"`
|
timeout time.Duration
|
}
|
|
type RequestSerfTopicResp struct {
|
DevId string `json:"devId"`
|
DevIp string `json:"devIp"`
|
Topic string `json:"topic"`
|
Procs []TopicProc `json:"procs"`
|
}
|
|
type TopicProc struct {
|
ProcId string `json:"procId"`
|
Addr bhome_msg.BHAddress `json:"addr"`
|
}
|
|
//通过topic获取指定topic的所有节点,供bus使用
|
func GetProcsInClusterByTopic(topic string) ([]RequestSerfTopicResp, error) {
|
req := RequestSerfTopicMsg{
|
Topic: QueryNodesByTopic,
|
timeout: time.Second * 10,
|
}
|
err, data := QuerySerfNodes(req)
|
if err != nil {
|
return nil, err
|
}
|
var result []RequestSerfTopicResp
|
if len(data) > 0 {
|
for _, d := range data {
|
var arr []RequestSerfTopicResp
|
if err = json.Unmarshal(d, &arr); err != nil {
|
logger.Debug("json Unmarshal err:", err)
|
} else {
|
result = append(result, arr...)
|
}
|
}
|
}
|
|
return result, nil
|
}
|
|
//响应根据topic获取地址
|
func ResponseGetNodeByTopic(topic string) ([]RequestSerfTopicResp, error) {
|
clients := hms.GetLocalNetNodeByTopic(topic)
|
if len(clients) == 0 {
|
return nil, errors.New("topic not exist")
|
} else {
|
ipv4, _, _ := util.GetLocalIP(config.Server.NetworkAdapter)
|
var destArr []TopicProc
|
for idx, c := range clients {
|
tp := TopicProc{
|
ProcId: string(c.ProcId),
|
Addr: *(clients[idx].Addr),
|
}
|
//if c. != nil {
|
// for _,p := range cli.Service.TopicList {
|
// if topic == string(p) {
|
// tps = append(tps, TopicProc{
|
// ProcId: string(cli.Proc.ProcId),
|
// ProcName: string(cli.Proc.Name),
|
// })
|
// }
|
// }
|
//}
|
destArr = append(destArr, tp)
|
}
|
r := append([]RequestSerfTopicResp{}, RequestSerfTopicResp{
|
Topic: topic,
|
DevId: config.Server.AnalyServerId,
|
DevIp: ipv4,
|
Procs: destArr,
|
})
|
return r, nil
|
}
|
}
|
|
//代理请求，通过当前的serf发送给其他的serf节点
|
func QuerySerfNodes(req RequestSerfTopicMsg) (error, [][]byte) {
|
resp, err := Agent.Query(req.Topic, req.Msg, &serf.QueryParam{
|
FilterNodes: req.targetNodeIds,
|
})
|
if err != nil {
|
return err, nil
|
}
|
|
var data [][]byte
|
var retE error
|
after := time.After(req.timeout)
|
loop:
|
for {
|
select {
|
case r := <-resp.ResponseCh():
|
data = append(data, r.Payload)
|
case <-after:
|
//retE = errors.New("time out")
|
break loop
|
}
|
}
|
|
return retE, data
|
}
|
|
//serf接收到query消息后需要访问本地进程获得结果并反馈
|
func QueryLocalProc(req RequestSerfTopicMsg) (error, interface{}) {
|
br := bhomeclient.Request{
|
Path: req.Topic,
|
Body: req.Msg,
|
}
|
reply, err := hms.RequestTopic(config.Server.AnalyServerId, br, 5000)
|
if err != nil {
|
return err, nil
|
}
|
return nil, reply
|
}
|
|
var (
|
riPool map[string]bhome_msg_dev.MsgDevRegisterInfo
|
topicPubAllRegisterInfo = "pub-allRegisterInfo-to-center"
|
riLock sync.Mutex
|
)
|
|
func init() {
|
riPool = make(map[string]bhome_msg_dev.MsgDevRegisterInfo)
|
}
|
|
//从serf中获取注册中心的注册信息是否变化，并将集群中所有注册信息汇总
|
func DoSyncRegisterInfo() error {
|
if hms == nil || Agent == nil {
|
return errors.New("ms or Agent handle is nil")
|
}
|
infos, err := hms.GetRegisteredClient()
|
if err != nil {
|
return err
|
}
|
ipv4, _, _ := util.GetLocalIP(config.Server.NetworkAdapter)
|
ri := bhome_msg_dev.MsgDevRegisterInfo{
|
DevId: []byte(config.Server.AnalyServerId),
|
Ip: []byte(ipv4),
|
ProxyPort: 4020,
|
}
|
for _, i := range infos {
|
if i.Online {
|
mta := bhome_msg_dev.MsgProcTopicsAll{
|
Online: i.Online,
|
}
|
if i.Proc != nil {
|
mta.Proc = &bhome_msg_dev.ProcInfo{
|
ProcId: i.Proc.ProcId,
|
Name: i.Proc.Name,
|
PublicInfo: i.Proc.PublicInfo,
|
PrivateInfo: i.Proc.PrivateInfo,
|
}
|
//proc的端口信息放在PublicInfo中
|
if string(mta.Proc.PublicInfo) != "" {
|
if iPort, e := strconv.Atoi(string(mta.Proc.PublicInfo)); e == nil && iPort > 0 && iPort < 65535 {
|
mta.Port = int32(iPort)
|
}
|
}
|
}
|
if i.Topics != nil {
|
mta.TopicList = &bhome_msg_dev.MsgTopicList{
|
TopicList: i.Topics.TopicList,
|
}
|
}
|
if i.SubLocalTopics != nil {
|
mta.LocalSublist = &bhome_msg_dev.MsgTopicList{
|
TopicList: i.SubLocalTopics.TopicList,
|
}
|
}
|
if i.SubNetTopics != nil {
|
mta.NetSublist = &bhome_msg_dev.MsgTopicList{
|
TopicList: i.SubNetTopics.TopicList,
|
}
|
}
|
ri.ProcInfos = append(ri.ProcInfos, &mta)
|
}
|
}
|
|
riLock.Lock()
|
bs := false //判断是否更改，如果发生更改则需要重新发布到center中
|
if v, ok := riPool[string(ri.DevId)]; !ok {
|
bs = true
|
riPool[string(ri.DevId)] = ri
|
} else {
|
if !reflect.DeepEqual(v, ri) {
|
bs = true
|
riPool[string(ri.DevId)] = ri
|
}
|
}
|
riLock.Unlock()
|
|
if bs {
|
bts, err := proto.Marshal(&ri)
|
if err != nil {
|
logger.Debug("marshal riPool err:", err)
|
return err
|
}
|
|
logger.Debug("UserEventSyncRegisterInfo len(bts):", len(bts))
|
Agent.UserEvent(UserEventSyncRegisterInfo, bts, false)
|
}
|
return nil
|
}
|
|
func compareRPool(ri *bhome_msg_dev.MsgDevRegisterInfo) {
|
logger.Debug("Enter compareRPool")
|
riLock.Lock()
|
riPool[string(ri.DevId)] = *ri
|
riLock.Unlock()
|
|
b, _ := proto.Marshal(ri)
|
|
logger.Debug("before hms.Publish to topicPubAllRegisterInfo")
|
err := hms.Publish(topicPubAllRegisterInfo, b)
|
if err != nil {
|
logger.Error("hms.Publish to topicPubAllRegisterInfo err:", err)
|
} else {
|
logger.Debug("hms.Publish success")
|
}
|
}
|
|
func SyncProxy(topic string, body []byte) error {
|
return Agent.UserEvent(topic, body, false)
|
}
|
