feat: 增加流式RPC测试代码

.gitignore

-example/plugin_rpc/client/client
-example/plugin_rpc/server/server
-example/plugin_mgr/plugin_mgr

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/pluginx.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module version="4">
+  <component name="Go" enabled="true" />
+</module>
\ No newline at end of file

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

example/plugin_rpc/client/client.go

@@ -22,6 +22,7 @@ func main() {
 
 	call(client)
 	sendFile(client)
+	chat(client)
 }
 
 func call(client *pluginrpc.PluginGrpcClient) {
@@ -54,3 +55,20 @@ func sendFile(client *pluginrpc.PluginGrpcClient) {
 	l.Info("SendFile res header:", res.Header)
 	l.Infof("SendFile res code:%d, desc:%s, data:%s", res.Code, res.Desc, string(res.Data))
 }
+
+func chat(client *pluginrpc.PluginGrpcClient) {
+	req := pluginrpc.NewReq("agent", "test-plugin", "chat", []byte("chat"))
+	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)
+	}
+}

example/plugin_rpc/server/server.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"time"
 
 	"linkfog.com/pluginx/pluginrpc"
 	pb "linkfog.com/pluginx/proto"
@@ -31,6 +32,16 @@ func SendFileHandle(fs *pb.FileStream, filePath string) {
 	l.Infof("recv file name:%s, purpose:%s, size:%d, path:%s", fs.Name, fs.Purpose, fs.TotalSize, filePath)
 }
 
+func (s *pluginServer) Chat(req *pb.Req, stream pb.Plugin_ChatServer) error {
+	moduleName := req.GetHeader().GetTo()
+	l.Infof("RPC:start chat with %s", moduleName)
+
+	ch = make(chan *pb.Res, 50)
+	return pluginrpc.ChatHelper(req, stream, ch)
+}
+
+var ch chan *pb.Res
+
 func main() {
 	pluginServer := newServer()
 	grpcServerHelper, err := pluginrpc.NewPluginGrpcServer("/tmp/unix_domain.sock", pluginServer)
@@ -40,5 +51,23 @@ func main() {
 	}
 	defer grpcServerHelper.Close()
 
+	time.Sleep(10 * time.Second)
+	go func() {
+		msg := &pb.Res{Header: &pb.Header{
+			UUID:      "123456",
+			Timestamp: time.Now().Unix(),
+			From:      "test-plugin",
+			To:        "agent",
+			Func:      "",
+		}, Code: 0, Desc: "chat info",
+		}
+		for {
+			msg.Header.Timestamp = time.Now().Unix()
+			msg.Code += 1
+			ch <- msg
+			time.Sleep(5 * time.Second)
+		}
+	}()
+
 	select {}
 }

pluginrpc/plugin_grpc_server.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"sync"
 	"time"
 
 	"google.golang.org/grpc"
@@ -119,3 +120,33 @@ func SendFileHelper(stream pb.Plugin_SendFileServer, callback SendFileCallBack)
 	res.Desc = "success"
 	return stream.SendAndClose(res)
 }
+
+func ChatHelper(msg *pb.Req, stream pb.Plugin_ChatServer, res <-chan *pb.Res) error {
+
+	moduleName := msg.GetHeader().GetTo()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stream.Context().Done():
+				l.Info("stream closed", moduleName)
+				return
+			case sig, ok := <-res:
+				if !ok {
+					l.Warn("sig chan closed", moduleName)
+					return
+				}
+				l.Infof(" send message: %v", msg)
+				if err := stream.Send(sig); err != nil {
+					l.Error(err)
+					return
+				}
+			}
+		}
+	}()
+
+	wg.Wait()
+	return nil
+}