[ Cmii ] [ Octopus ] - reformat agent-go - 1

agent-go/rabbitmq/RabbitMsgQueue.go

@@ -1,17 +1,18 @@
 package rabbitmq
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/streadway/amqp"
 	"strings"
 	"sync"
 	"wdd.io/agent-common/logger"
-	"wdd.io/agent-go/g"
 )
 
 type RabbitMQ interface {
 	RabbitSendWriter
 	RabbitConnectCloser
+	RabbitQueueHandler
 }
 
 type RabbitSendWriter interface {
@@ -26,10 +27,19 @@ type RabbitConnectCloser interface {
 	Close() error
 }
 
-type RabbitQueue struct {
-	RabbitConn *RabbitMQConn
+type RabbitQueueHandler interface {
+	Handle() chan bool
+}
 
+type RabbitQueue struct {
+	// 连接实体
+	RabbitConn *RabbitMQConn
+	// 连接属性
 	RabbitProp *ConnectProperty
+	// 底层连接tcp信息
+	RabbitConnectInfo RabbitTCPConnectInfo
+	// 返回消息队列
+	ReceiveChan *RabbitReceiveChan
 }
 
 // RabbitMQConn is a struct that holds the connection and channel objects
@@ -38,6 +48,13 @@ type RabbitMQConn struct {
 	Channel    *amqp.Channel
 }
 
+type RabbitReceiveChan struct {
+	AgentRChan    chan *OctopusMessage
+	ExecutorRChan chan *OctopusMessage
+	StatusRChan   chan *OctopusMessage
+	InitRChan     chan *OctopusMessage
+}
+
 type ConnectProperty struct {
 	ExchangeName string
 	QueueName    string
@@ -45,6 +62,14 @@ type ConnectProperty struct {
 	TopicKey     string
 }
 
+type RabbitTCPConnectInfo struct {
+	UserName    string
+	Password    string
+	Host        string
+	Port        string
+	VirtualHost string
+}
+
 var log = logger.Log
 
 // 定义全局唯一的 Singleton 实例
@@ -54,25 +79,28 @@ var instance *amqp.Connection
 var once sync.Once
 
 // 初始化 Singleton 实例的函数
-func createInstance() {
+func createInstance(rabbitConnectInfo RabbitTCPConnectInfo) func() {
 	// 在这里进行 Singleton 的初始化操作
 
 	// 获取RabbitMQ的连接地址
-	rabbitMQEndpointFromG := parseRabbitMQEndpointFromG()
+	rabbitMQEndpoint := parseRabbitMQEndpoint(rabbitConnectInfo)
 
 	// 创建全局唯一连接 RabbitMQ连接
-	connection, err := amqp.Dial(rabbitMQEndpointFromG)
+	connection, err := amqp.Dial(rabbitMQEndpoint)
 	if err != nil {
 		log.Error(fmt.Sprintf("failed to connect to RabbitMQ: %v", err))
 	}
 
 	instance = connection
+	return nil
 }
 
 // GetInstance 获取全局唯一的 Singleton 实例的函数
-func GetInstance() *amqp.Connection {
+func GetInstance(rabbitConnectInfo RabbitTCPConnectInfo) *amqp.Connection {
 	// 使用 sync.Once 确保 createInstance 只会被调用一次
-	once.Do(createInstance)
+
+	// todo 理解
+	once.Do(createInstance(rabbitConnectInfo))
 	return instance
 }
 
@@ -80,7 +108,7 @@ func GetInstance() *amqp.Connection {
 func (r *RabbitQueue) Connect() {
 
 	// 获取RabbitMQ的连接
-	conn := GetInstance()
+	conn := GetInstance(r.RabbitConnectInfo)
 
 	ch, err := conn.Channel()
 	if err != nil {
@@ -121,10 +149,26 @@ func (r *RabbitQueue) Connect() {
 		log.Error(fmt.Sprintf("failed to bind RabbitMQ queue: %w", err))
 	}
 
+	// build for receive chan
+	rabbitRCha := &RabbitReceiveChan{}
+	if strings.HasPrefix(r.RabbitProp.QueueName, "Init") {
+		// init queue
+		rabbitRCha.InitRChan = make(chan *OctopusMessage)
+	} else {
+		// business queue
+		rabbitRCha.AgentRChan = make(chan *OctopusMessage, 5)
+		rabbitRCha.ExecutorRChan = make(chan *OctopusMessage, 5)
+		rabbitRCha.StatusRChan = make(chan *OctopusMessage, 5)
+	}
+
+	// connection
 	r.RabbitConn = &RabbitMQConn{
 		Connection: conn,
 		Channel:    ch,
 	}
+
+	// receive chan
+	r.ReceiveChan = rabbitRCha
 }
 
 func (r *RabbitQueue) Close() error {
@@ -138,6 +182,29 @@ func (r *RabbitQueue) Close() error {
 	return err
 }
 
+func (r *RabbitQueue) Handle() chan bool {
+
+	deliveries := r.Read(true)
+	forverHandle := make(chan bool)
+	// 死循环，处理Octopus Message
+	for delivery := range deliveries {
+
+		var om *OctopusMessage
+		err := json.Unmarshal(delivery.Body, &om)
+		if err != nil {
+			log.Error(fmt.Sprintf("octopus message convert to json is wrong! msg is => %s", delivery.Body))
+			// 保存到某处
+			continue
+		}
+		// 策略模式 处理消息
+		P.Submit(func() {
+			om.HandleMsg(r.ReceiveChan)
+		})
+	}
+
+	return forverHandle
+}
+
 // Send 向RabbitMQ中发送消息
 func (r *RabbitQueue) Send(message []byte) {
 	// 往哪里发
@@ -159,6 +226,15 @@ func (r *RabbitQueue) Send(message []byte) {
 	}
 }
 
+func (r *RabbitQueue) SendOMsg(oMessage *OctopusMessage) {
+	bytes, err := json.Marshal(&oMessage)
+	if err != nil {
+		log.ErrorF("octopus message Marshal error %v", &oMessage)
+	}
+
+	r.Send(bytes)
+}
+
 func (r *RabbitQueue) Read(autoAck bool) <-chan amqp.Delivery {
 
 	// 拿到特定的Channel
@@ -181,30 +257,21 @@ func (r *RabbitQueue) Read(autoAck bool) <-chan amqp.Delivery {
 	return msgs
 }
 
-// parseRabbitMQEndpoint 根据全局变量agentConfig解析出RabbitMQ的连接地址
-func parseRabbitMQEndpointFromG() string {
-
-	agentConfig := g.G.AgentConfig
+func parseRabbitMQEndpoint(rabbitConnectInfo RabbitTCPConnectInfo) string {
 
 	var res strings.Builder
 
-	host := agentConfig.GetString("spring.rabbitmq.host")
-	port := agentConfig.GetString("spring.rabbitmq.port")
-	username := agentConfig.GetString("spring.rabbitmq.username")
-	password := agentConfig.GetString("spring.rabbitmq.password")
-	virtualHost := agentConfig.GetString("spring.rabbitmq.virtual-host")
-
 	// amqp://{username}:{password}@{hostname}:{port}/{virtual_host}
 	res.WriteString("amqp://")
-	res.WriteString(username)
+	res.WriteString(rabbitConnectInfo.UserName)
 	res.WriteString(":")
-	res.WriteString(password)
+	res.WriteString(rabbitConnectInfo.Password)
 	res.WriteString("@")
-	res.WriteString(host)
+	res.WriteString(rabbitConnectInfo.Host)
 	res.WriteString(":")
-	res.WriteString(port)
+	res.WriteString(rabbitConnectInfo.Port)
 	res.WriteString("/")
-	res.WriteString(virtualHost)
+	res.WriteString(rabbitConnectInfo.VirtualHost)
 
 	s := res.String()