nodehub/example/echo/client.go at master · joyparty/nodehub · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
package main

import (
	"context"
	"crypto/tls"
	"flag"
	"fmt"
	"log/slog"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/joyparty/gokit"
	"github.com/joyparty/nodehub/component/gateway/client"
	"github.com/joyparty/nodehub/example/echo/proto/authpb"
	"github.com/joyparty/nodehub/example/echo/proto/clusterpb"
	"github.com/joyparty/nodehub/example/echo/proto/echopb"
	"github.com/joyparty/nodehub/logger"
	"google.golang.org/grpc/status"
)

var (
	serverAddr string
	useTCP     bool
	useQUIC    bool

	echoServiceCode = int32(clusterpb.Services_ECHO)
	authServiceCode = int32(clusterpb.Services_AUTH)
)

func init() {
	flag.StringVar(&serverAddr, "server", "127.0.0.1:9000", "server address")
	flag.BoolVar(&useTCP, "tcp", false, "use tcp")
	flag.BoolVar(&useQUIC, "quic", false, "use quic")
	flag.Parse()

	logger.SetLogger(
		slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
			Level: slog.LevelDebug,
		})),
	)
}

func main() {
	var endpoint string
	if useTCP {
		endpoint = fmt.Sprintf("tcp://%s", serverAddr)
	} else if useQUIC {
		endpoint = fmt.Sprintf("quic://%s", serverAddr)
	} else {
		endpoint = fmt.Sprintf("ws://%s/", serverAddr)
	}

	cli := newClient(endpoint)
	defer cli.Close()

	// 鉴权请求
	authReply := &authpb.AuthorizeAck{}
	err := cli.Call(context.Background(), 0, "Authorize",
		authpb.AuthorizeToken_builder{
			Token: "0d8b750e-35e8-4f98-b032-f389d401213e",
		}.Build(),
		authReply,
	)
	if err != nil {
		handleError(err)
	}
	logger.Info("auth success", "user_id", authReply.GetUserId())

	var wg sync.WaitGroup
	for {

		for i := 0; i < 3; i++ {
			wg.Add(1)

			go func() {
				defer wg.Done()

				echoReply := &echopb.Msg{}
				err := cli.Call(context.Background(), echoServiceCode, "Send",
					echopb.Msg_builder{Content: "hello world!"}.Build(),
					echoReply,
					client.WithStream("test:echo"),
				)

				if err != nil {
					handleError(err)
				} else {
					logger.Info("echo back", "content", echoReply.GetContent())
				}
			}()
		}

		wg.Wait()
		time.Sleep(1 * time.Second)
	}
}

func handleError(err error) {
	if s, ok := status.FromError(err); ok {
		fmt.Printf("RPCError, code = %s, message = %s\n", s.Code(), s.Message())
	} else {
		fmt.Printf("ERROR: %v\n", err)
	}

	os.Exit(1)
}

func newClient(endpoint string) *client.Client {
	var cli *client.Client
	if strings.HasPrefix(endpoint, "quic://") {
		tlsConfig := &tls.Config{
			InsecureSkipVerify: true,
			NextProtos:         []string{"quic-echo-example"},
		}
		cli = gokit.MustReturn(client.NewQUIC(endpoint, tlsConfig, nil))
	} else {
		cli = gokit.MustReturn(client.New(endpoint))
	}

	return cli
}