101 lines
2.1 KiB
Go
101 lines
2.1 KiB
Go
package mq
|
|
|
|
import (
|
|
"fmt"
|
|
"git.hpds.cc/Component/logging"
|
|
"git.hpds.cc/Component/network/frame"
|
|
"go.uber.org/zap"
|
|
"os"
|
|
"taskExecute/config"
|
|
"time"
|
|
|
|
"git.hpds.cc/pavement/hpds_node"
|
|
)
|
|
|
|
var MqList []HpdsMqNode
|
|
|
|
type HpdsMqNode struct {
|
|
MqType uint
|
|
Topic string
|
|
Node config.HpdsNode
|
|
EndPoint interface{}
|
|
Logger *logging.Logger
|
|
}
|
|
|
|
func must(logger *logging.Logger, err error) {
|
|
if err != nil {
|
|
if logger != nil {
|
|
logger.With(zap.String("任务执行节点", "错误信息")).Error("启动错误", zap.Error(err))
|
|
} else {
|
|
_, _ = fmt.Fprint(os.Stderr, err)
|
|
}
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *logging.Logger) (mqList []HpdsMqNode, err error) {
|
|
mqList = make([]HpdsMqNode, 0)
|
|
for _, v := range funcs {
|
|
switch v.MqType {
|
|
case 2:
|
|
sf := hpds_node.NewStreamFunction(
|
|
v.Name,
|
|
hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", node.Host, node.Port)),
|
|
hpds_node.WithObserveDataTags(frame.Tag(v.DataTag)),
|
|
hpds_node.WithCredential(node.Token),
|
|
)
|
|
err = sf.Connect()
|
|
must(logger, err)
|
|
nodeInfo := HpdsMqNode{
|
|
MqType: 2,
|
|
Topic: v.Name,
|
|
Node: node,
|
|
EndPoint: sf,
|
|
}
|
|
switch v.Name {
|
|
case "task-execute":
|
|
_ = sf.SetHandler(TaskExecuteHandler)
|
|
default:
|
|
|
|
}
|
|
mqList = append(mqList, nodeInfo)
|
|
default:
|
|
ap := hpds_node.NewAccessPoint(
|
|
v.Name,
|
|
hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", node.Host, node.Port)),
|
|
hpds_node.WithCredential(node.Token),
|
|
)
|
|
err = ap.Connect()
|
|
nodeInfo := HpdsMqNode{
|
|
MqType: 1,
|
|
Topic: v.Name,
|
|
Node: node,
|
|
EndPoint: ap,
|
|
}
|
|
must(logger, err)
|
|
ap.SetDataTag(frame.Tag(v.DataTag))
|
|
mqList = append(mqList, nodeInfo)
|
|
}
|
|
|
|
}
|
|
return mqList, err
|
|
}
|
|
|
|
func GetMqClient(topic string, mqType uint) *HpdsMqNode {
|
|
for _, v := range MqList {
|
|
if v.Topic == topic && v.MqType == mqType {
|
|
return &v
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func GenerateAndSendData(stream hpds_node.AccessPoint, data []byte) error {
|
|
_, err := stream.Write(data)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
time.Sleep(1000 * time.Millisecond)
|
|
return nil
|
|
}
|