[agent-go] 初步完成Executor部分的代码- 1

agent-go/IRabbitSendWriter.go

@@ -0,0 +1,11 @@
+package main
+
+/*type RabbitSendWriter interface {
+
+	Send(conn *RabbitMQConn, connProp *ConnectProperty, message []byte)
+
+	Read(conn *RabbitMQConn, connProp *ConnectProperty, autoAck bool) <-chan amqp.Delivery
+
+}
+
+*/

agent-go/config/AgentServerInfo.go

@@ -1,4 +1,4 @@
-package register
+package config
 
 type AgentServerInfo struct {
 	ServerName     string `json:"serverName" yaml:"serverName"`

agent-go/config/OctopusMessage.go

@@ -1,7 +1,7 @@
-package rabbitmq
+package config
 
 import (
-	"agent-go/g"
+	"agent-go/utils"
 	"time"
 )
 
@@ -28,7 +28,7 @@ type ExecutionMessage struct {
 func (m *OctopusMessage) BuildOctopusMsg(omType string, content interface{}) *OctopusMessage {
 
 	// 当前时间
-	curTimeString := g.CurTimeString()
+	curTimeString := utils.CurTimeString()
 
 	return &OctopusMessage{
 		UUID:     curTimeString,

agent-go/executor/CommandExecutor.go

@@ -1,8 +1,8 @@
 package executor
 
 import (
+	"agent-go/config"
 	"agent-go/g"
-	"agent-go/rabbitmq"
 	"bufio"
 	"bytes"
 	"fmt"
@@ -11,7 +11,7 @@ import (
 
 var log = g.G.LOG
 
-func Execute(om *rabbitmq.OctopusMessage, em *rabbitmq.ExecutionMessage) ([]string, error) {
+func Execute(om *config.OctopusMessage, em *config.ExecutionMessage) ([]string, error) {
 
 	var resultLog []string
 	var err error

agent-go/g/NacosConfig.go

@@ -1,7 +1,6 @@
-package config
+package g
 
 import (
-	"agent-go/g"
 	"bytes"
 	"fmt"
 	"github.com/nacos-group/nacos-sdk-go/v2/clients"
@@ -14,7 +13,7 @@ import (
 	"strings"
 )
 
-var log = g.G.LOG
+var log = G.LOG
 var group = ""
 
 func InitNacos(configFileName string) {

agent-go/g/global.go

@@ -1,14 +1,14 @@
 package g
 
 import (
-	"agent-go/register"
+	"agent-go/config"
 	"github.com/spf13/viper"
 )
 
 type Global struct {
 	LOG             *Logger
 	NacosConfig     *viper.Viper
-	AgentServerInfo *register.AgentServerInfo
+	AgentServerInfo *config.AgentServerInfo
 }
 
 const (
@@ -17,6 +17,8 @@ const (
 	ExecOmType   = "EXECUTOR"
 	StatusOmType = "STATUS"
 	InitOmType   = "INIT"
+
+	// write about
 )
 
 var logger, _ = NewLogger()

agent-go/main.go

@@ -1,11 +1,14 @@
 package main
 
 import (
+	"agent-go/g"
 	"agent-go/register"
 	"flag"
 	"fmt"
 )
 
+var log = g.G.LOG
+
 func main() {
 
 	// 解析命令行参数
@@ -20,6 +23,6 @@ func main() {
 	//config.InitNacos(filename)
 
 	// 执行初始化之策工作
-	register.INIT()
+	g.G.AgentServerInfo = register.INIT()
 
 }

agent-go/rabbitmq/MessageReaderWriter.go

@@ -6,6 +6,19 @@ import (
 	"github.com/streadway/amqp"
 )
 
+// RabbitMQConn is a struct that holds the connection and channel objects
+type RabbitMQConn struct {
+	Connection *amqp.Connection
+	Channel    *amqp.Channel
+}
+
+type ConnectProperty struct {
+	ExchangeName string
+	QueueName    string
+	ExchangeType string
+	TopicKey     string
+}
+
 // Send 向RabbitMQ中发送消息
 func Send(conn *RabbitMQConn, connProp *ConnectProperty, message []byte) {
 	// 往哪里发

agent-go/rabbitmq/OctopusMsgHandler.go

@@ -1,13 +1,14 @@
 package rabbitmq
 
 import (
+	"agent-go/config"
 	"agent-go/executor"
 	"agent-go/g"
 	"encoding/json"
 	"fmt"
 )
 
-func HandleOMsg(initOMsgFromServer *OctopusMessage) {
+func HandleOMsg(initOMsgFromServer *config.OctopusMessage) {
 
 	agentTopicName := initOMsgFromServer.UUID
 	OctopusExchange := g.G.NacosConfig.GetString("octopus.message.octopus_exchange")
@@ -43,7 +44,7 @@ func HandleOMsg(initOMsgFromServer *OctopusMessage) {
 	// 死循环，处理Ocotpus Message
 	for delivery := range deliveries {
 
-		var om *OctopusMessage
+		var om *config.OctopusMessage
 		err := json.Unmarshal(delivery.Body, &om)
 		if err != nil {
 			log.Error("Octopus Message Parse Error !")
@@ -57,7 +58,7 @@ func HandleOMsg(initOMsgFromServer *OctopusMessage) {
 
 }
 
-func doHandleOctopusMessage(octopusMessage *OctopusMessage) {
+func doHandleOctopusMessage(octopusMessage *config.OctopusMessage) {
 
 	switch octopusMessage.Type {
 	case g.InitOmType:
@@ -72,11 +73,11 @@ func doHandleOctopusMessage(octopusMessage *OctopusMessage) {
 
 }
 
-func executorOMHandler(octopusMessage *OctopusMessage) {
+func executorOMHandler(octopusMessage *config.OctopusMessage) {
 
 	executionMsgString := octopusMessage.Content.(string)
 
-	var executionMessage *ExecutionMessage
+	var executionMessage *config.ExecutionMessage
 	err := json.Unmarshal([]byte(executionMsgString), &executionMessage)
 	if err != nil {
 		return
@@ -87,10 +88,10 @@ func executorOMHandler(octopusMessage *OctopusMessage) {
 
 }
 
-func statusOMHandler(octopusMessage *OctopusMessage) {
+func statusOMHandler(octopusMessage *config.OctopusMessage) {
 
 }
 
-func blackHoleOMHandler(octopusMessage *OctopusMessage) {
+func blackHoleOMHandler(octopusMessage *config.OctopusMessage) {
 	log.Error(fmt.Sprintf("octopusMessage type wrong! msg is => %v", octopusMessage))
 }

agent-go/rabbitmq/RabbitMQConnector.go

@@ -10,19 +10,6 @@ import (
 
 var log = g.G.LOG
 
-// RabbitMQConn is a struct that holds the connection and channel objects
-type RabbitMQConn struct {
-	Connection *amqp.Connection
-	Channel    *amqp.Channel
-}
-
-type ConnectProperty struct {
-	ExchangeName string
-	QueueName    string
-	ExchangeType string
-	TopicKey     string
-}
-
 // 定义全局唯一的 Singleton 实例
 var instance *amqp.Connection
 

agent-go/register/AgentIntitilization.go

@@ -1,6 +1,7 @@
 package register
 
 import (
+	"agent-go/config"
 	"agent-go/g"
 	"agent-go/rabbitmq"
 	"encoding/json"
@@ -9,14 +10,13 @@ import (
 	"io/ioutil"
 )
 
-var log = g.G.LOG
 var omType = g.InitOmType
-var agentServerInfo = g.G.AgentServerInfo
+var log = g.G.LOG
 
-func INIT() {
+func INIT() *config.AgentServerInfo {
 
 	// 获取系统的环境变量
-	g.G.AgentServerInfo = parseAgentServerInfo()
+	agentServerInfo := parseAgentServerInfo()
 
 	nacosConfig := g.G.NacosConfig
 
@@ -37,6 +37,7 @@ func INIT() {
 	// 建立RabbitMQ的连接
 	// defer 关闭初始化连接
 	initToServer, err := rabbitmq.NewRabbitMQConn(
+
 		initToServerProp,
 	)
 	if err != nil {
@@ -55,7 +56,7 @@ func INIT() {
 	defer rabbitmq.CloseChannel(initFromServer)
 
 	// 组装OctopusMessage
-	var octopusMsg *rabbitmq.OctopusMessage
+	var octopusMsg *config.OctopusMessage
 	octopusMsg = octopusMsg.BuildOctopusMsg(
 		omType,
 		agentServerInfo,
@@ -75,6 +76,8 @@ func INIT() {
 	// 建立运行时RabbitMQ连接
 	handleInitMsgFromServer(initFromServer, initFromServerProp, initToServer, initToServerProp)
 
+	return agentServerInfo
+
 }
 
 // handleInitMsgFromServer 处理从Server接收的注册信息
@@ -85,7 +88,7 @@ func handleInitMsgFromServer(initFromServer *rabbitmq.RabbitMQConn, initFromServ
 	// 同步很多抢占注册的情况
 	for delivery := range deliveries {
 
-		var om *rabbitmq.OctopusMessage
+		var om *config.OctopusMessage
 		err := json.Unmarshal(delivery.Body, &om)
 		if err != nil {
 			log.Error(fmt.Sprintf("parse init message from server wroong, message is =>  %s ",
@@ -120,11 +123,11 @@ func shutdownRegisterQueueConnection(initFromServer *rabbitmq.RabbitMQConn, init
 
 }
 
-func parseAgentServerInfo() *AgentServerInfo {
+func parseAgentServerInfo() *config.AgentServerInfo {
 
 	// 约定文件地址为 /etc/environment.d/octopus-agent.conf
 	// 目前使用
-	var agentServerInfo *AgentServerInfo
+	var agentServerInfo *config.AgentServerInfo
 	yamlFile, err := ioutil.ReadFile("C:\\Users\\wdd\\IdeaProjects\\ProjectOctopus\\agent-go\\server-env.yaml")
 	if err != nil {
 		panic(fmt.Errorf("failed to read YAML file: %v", err))

agent-go/utils/TimeUtils.go

@@ -1,4 +1,4 @@
-package g
+package utils
 
 import (
 	"time"