add

Author: caoyingjunz

practise/grpc/client.go

@@ -1,11 +1,9 @@
 package main
 
 import (
-	"bufio"
 	"context"
 
 	"log"
-	"os"
 	"time"
 
 	"google.golang.org/grpc"
@@ -32,6 +30,9 @@ func main() {
 		log.Fatalf("%v", err)
 	}
 
+	clientId := "node2"
+
+	// 启动协程，接受服务段回调 client 的请求
 	go func() {
 		for {
 			msg, err := stream.Recv()
@@ -43,15 +44,17 @@ func main() {
 		}
 	}()
 
-	reader := bufio.NewReader(os.Stdin)
-	for {
-		text, _ := reader.ReadString('\n')
-		if err := stream.Send(&pd.Request{
-			Type:    "server_call",
-			Payload: []byte(text),
+	// 启动客户端定时测试DEMO
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		ts := time.Now().String()
+		if err = stream.Send(&pd.Request{
+			Type:    clientId,
+			Payload: []byte(ts),
 		}); err != nil {
-			log.Fatalf("Send failed: %v", err)
+			log.Println("调用服务端失败", err)
 		}
-		time.Sleep(100 * time.Millisecond)
 	}
 }

practise/grpc/server.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"sync"
 
+	"github.com/gin-gonic/gin"
 	"google.golang.org/grpc"
 
 	pd "go-learning/practise/grpc/tunnel"
@@ -29,30 +30,32 @@ func (s *server) Connect(stream pd.Tunnel_ConnectServer) error {
 			log.Printf("tream.Recv %v", err)
 			return err
 		}
-		log.Printf("Received from %s: %s", req.Type, string(req.Payload))
-
-		switch req.Type {
-		case "client_call":
-			// 处理客户端主动调用
-			fmt.Println("client_call", req)
-		case "server_call":
-			// 这是 Server 发起的调用，Client 应返回结果
-			resp := &pd.Response{Result: []byte("server pong")}
-			if err = stream.Send(resp); err != nil {
-				return err
-			}
+
+		s.lock.Lock()
+		_, ok := s.clients[req.Type]
+		if !ok {
+			s.clients[req.Type] = stream
 		}
+		s.lock.Unlock()
+
+		// TODO 目前是DEMO
+		log.Printf("Received from %s %s", req.Type, string(req.Payload))
 	}
 }
+func (s *server) Call(c *gin.Context) {
+	_, _ = s.CallClient(c.Query("clientId"), nil)
+}
 
-func (s *server) CallClient(clientID string, data []byte) ([]byte, error) {
-	stream, ok := s.clients[clientID]
+func (s *server) CallClient(clientId string, data []byte) ([]byte, error) {
+	stream, ok := s.clients[clientId]
 	if !ok {
 		return nil, fmt.Errorf("client not connected")
 	}
 
 	// 发送调用请求
-	err := stream.Send(&pd.Response{})
+	err := stream.Send(&pd.Response{
+		Result: []byte(clientId + " server callback"),
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -66,11 +69,22 @@ func main() {
 		log.Fatalf("failed to listen %v", err)
 	}
 
+	cs := &server{clients: make(map[string]pd.Tunnel_ConnectServer)}
+
 	s := grpc.NewServer()
-	pd.RegisterTunnelServer(s, &server{})
+	pd.RegisterTunnelServer(s, cs)
+
+	go func() {
+		log.Printf("grpc listening at %v", listener.Addr())
+		if err = s.Serve(listener); err != nil {
+			log.Fatalf("failed to serve %v", err)
+		}
+	}()
 
-	log.Printf("listening at %v", listener.Addr())
-	if err = s.Serve(listener); err != nil {
-		log.Fatalf("failed to serve %v", err)
+	r := gin.Default()
+	r.GET("/ping", cs.Call)
+	log.Printf("http listening at %v", ":8093")
+	if err = r.Run(":8093"); err != nil {
+		log.Fatalf("failed to start http: %v", err)
 	}
 }