feat: 增加发布消息的处理

cmd/agent/global/mqtt.go

@@ -12,6 +12,8 @@ const (
 	ConsumerTopicPluginUpgrade    = "plugin-upgrade"
 	ConsumerTopicStartupPlugin    = "startup-plugin"
 	ConsumerTopicStopPlugin       = "stop-plugin"
+
+	PublishTopicReportInfo = "report-info"
 )
 
 var SubscribePrefixInfo = ""

go.mod

@@ -4,7 +4,7 @@ go 1.23.2
 
 replace linkfog.com/public => web.lueluesay.top/git/lil/public v0.0.0-20241017054747-f08bac907b6c
 
-replace linkfog.com/pluginx => web.lueluesay.top/git/lil/pluginx v0.0.0-20241017031415-3651ccc20390
+replace linkfog.com/pluginx => web.lueluesay.top/git/lil/pluginx v0.0.0-20241017112237-9affa4009a35
 
 require (
 	github.com/beevik/ntp v1.4.3

go.sum

@@ -60,7 +60,7 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntN
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-web.lueluesay.top/git/lil/pluginx v0.0.0-20241017031415-3651ccc20390 h1:V3uBqmiRc4xlBWfHRDETMGHTvLF+3mOGIZH/E46XcNI=
-web.lueluesay.top/git/lil/pluginx v0.0.0-20241017031415-3651ccc20390/go.mod h1:DL73qsfCIFfH2K6tdOLPj9tRcZ8AYxfPrkcTebYXr5o=
+web.lueluesay.top/git/lil/pluginx v0.0.0-20241017112237-9affa4009a35 h1:XgY8a7iuuwJqtzCzN50i9ZVzUhJFpp0BeaMAJgkvAj0=
+web.lueluesay.top/git/lil/pluginx v0.0.0-20241017112237-9affa4009a35/go.mod h1:DL73qsfCIFfH2K6tdOLPj9tRcZ8AYxfPrkcTebYXr5o=
 web.lueluesay.top/git/lil/public v0.0.0-20241017054747-f08bac907b6c h1:U/2ffmB20mPiC1GT0eTrc9GAQDCd2a8W4zAIb8wf1ds=
 web.lueluesay.top/git/lil/public v0.0.0-20241017054747-f08bac907b6c/go.mod h1:x/nRP9pMRVToI9Te1TazybP0Qlj3V+/aA2EiPQEvzsI=

module/plugin/config.go

@@ -10,6 +10,7 @@ import (
 
 // 插件名
 const (
+	AgentSelfPlugin     = "agent"
 	PushStreamingPlugin = "pushStreaming"
 	ReportDCSInfoPlugin = "reportDCSInfo"
 )

module/plugin/plugin.go

@@ -165,7 +165,7 @@ func (p *Plugin) dealWithSig(msg *global.Message) {
 	case global.ConsumerTopicPluginUpgrade:
 
 	case global.ConsumerTopicStartupPlugin:
-
+		go chatWithPlugin(p.grpcClient, ReportDCSInfoPlugin, global.PublishTopicReportInfo)
 	case global.ConsumerTopicStopPlugin:
 	}
 }

module/plugin/publish.go

+package plugin
+
+import (
+	"context"
+
+	"agent/cmd/agent/global"
+	"linkfog.com/pluginx/pluginrpc"
+	"linkfog.com/public/lib/l"
+)
+
+func chatWithPlugin(client *pluginrpc.PluginGrpcClient, pluginName string, topicName string) {
+	req := pluginrpc.NewReq(AgentSelfPlugin, pluginName, "", []byte{})
+	chatClient, err := client.Chat(context.Background(), req)
+	if err != nil {
+		l.Error("ChatClient err:", err)
+		return
+	}
+	for {
+		res, err := chatClient.Recv()
+		if err != nil {
+			l.Error("ChatClient receive err:", err)
+			return
+		}
+		l.Infof("receive message: %v", res)
+		global.HostInfoTail.MQTTInfo.PublishChan <- &global.Message{
+			Key:     topicName,
+			Payload: string(res.Data),
+		}
+	}
+}