新增基于grpc的双向通信DEMO (#113)

Author: caoyingjunz

practise/grpc/README.md

@@ -0,0 +1,83 @@
+# gRPC Usage
+
+### 目标
+基于grpc的双向通信demo
+
+### protoc (Protocol buffer compiler) 安装
+
+Install `protoc` from [protobuf](https://github.com/protocolbuffers/protobuf/releases)
+```shell
+protoc --version  # Ensure compiler version is 3+
+```
+
+Install `Go plugins` for the `protoc`
+
+```shell
+# Install the protocol compiler plugins for Go using the following commands:
+go install google.golang.org/protobuf/cmd/[email protected]
+go install google.golang.org/grpc/cmd/[email protected]
+
+# Update your PATH so that the protoc compiler can find the plugins:
+export PATH="$PATH:$(go env GOPATH)/bin"
+```
+
+### 构建 gRPC 服务
+
+定义 `.proto` 文件，本文以 `pixiu.proto` 为例
+
+```protobuf
+syntax="proto3";
+
+option go_package = "go-learning/practise/grpc-practise/tunnel/tunnel";
+
+package tunnel;
+
+service Tunnel {
+  // Client 调用此方法建立连接
+  rpc Connect(stream Request) returns (stream Response);
+}
+
+message Request {
+  string type = 1;  // "client_call" 或 "server_call"
+  bytes payload = 2;
+}
+
+message Response {
+  bytes result = 1;
+}
+```
+
+### 生成 `gRPC` 代码
+
+```shell
+protoc --go_out=. --go_opt=paths=source_relative \
+  --go-grpc_out=. --go-grpc_opt=paths=source_relative \
+  tunnel/tunnel.proto
+```
+
+执行命令之后，会在 `tunnel` 目录生成 `tunnel.pb.go` 和 `tunnel_grpc.pb.go` 代码文件
+- `tunnel.pb.go` 结构体
+- `tunnel_grpc.pb.go`: 客户端和服务端代码
+
+### 实现 gRPC 服务端
+```
+```
+
+### 实现 gRPC 客户端
+```
+```
+
+### 执行
+
+启动 `gRPC server`
+``` shell
+go run server.go
+```
+
+执行 `gRPC client`
+``` shell
+go run client.go
+
+# 回显
+2022/03/20 19:43:13 say hello caoyingjun 12345
+```

practise/grpc/client.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"context"
+
+	"log"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	pd "go-learning/practise/grpc/tunnel"
+)
+
+var (
+	addr = "127.0.0.1:8092"
+)
+
+func main() {
+	conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Fatalf("failed to connect rpc server %v", err)
+	}
+	defer conn.Close()
+
+	c := pd.NewTunnelClient(conn)
+
+	stream, err := c.Connect(context.Background())
+	if err != nil {
+		log.Fatalf("%v", err)
+	}
+
+	clientId := "node2"
+
+	// 启动协程，接受服务段回调 client 的请求
+	go func() {
+		for {
+			msg, err := stream.Recv()
+			if err != nil {
+				log.Printf("Receive error: %v", err)
+				return
+			}
+			log.Printf("Received from server: %s", msg.Result)
+		}
+	}()
+
+	// 启动客户端定时测试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.Println("调用服务端失败", err)
+		}
+	}
+}

practise/grpc/server.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"sync"
+
+	"github.com/gin-gonic/gin"
+	"google.golang.org/grpc"
+
+	pd "go-learning/practise/grpc/tunnel"
+)
+
+type server struct {
+	pd.UnimplementedTunnelServer
+
+	clients map[string]pd.Tunnel_ConnectServer
+	lock    sync.RWMutex
+}
+
+func (s *server) Connect(stream pd.Tunnel_ConnectServer) error {
+	for {
+		req, err := stream.Recv()
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			log.Printf("tream.Recv %v", err)
+			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]
+	if !ok {
+		return nil, fmt.Errorf("client not connected")
+	}
+
+	// 发送调用请求
+	err := stream.Send(&pd.Response{
+		Result: []byte(clientId + " server callback"),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return nil, err
+}
+
+func main() {
+	listener, err := net.Listen("tcp", ":8092")
+	if err != nil {
+		log.Fatalf("failed to listen %v", err)
+	}
+
+	cs := &server{clients: make(map[string]pd.Tunnel_ConnectServer)}
+
+	s := grpc.NewServer()
+	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)
+		}
+	}()
+
+	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)
+	}
+}