feat： rpc client service is ready, add direct builder and discovery builder

rpc 客户端服务，添加直连 rpc 逻辑，和服务发现模式

Author: coderi421

.golangci.yml

@@ -1,6 +1,6 @@
 run:
   timeout: 5m
-  modules-download-mode: readonly
+  modules-download-mode: mod
 
   govet:
     disable-all: true

README.md

@@ -73,18 +73,22 @@ Microservice framework implemented based on Golang.
 │      │  ├─pprof
 │      │  └─validation
 │      └─rpcserver
-│          ├─clientinterceptors
-│          ├─resolver
-│          │  ├─direct
-│          │  └─discovery
+│          ├─client.go rpc 客户端的初始化配置
+│          ├─server.go rpc 服务端的初始化配置
+│          ├─clientinterceptors 客户端的拦截器：超时连接器
+│          ├─resolver 服务发现相关的逻辑 解析器
+│          │  ├─direct 直连
+│          │  └─discovery 服务发现，负载均衡
+│          │      ├─builder.go 服务发现的构建器
+│          │      └─resolver.go 服务发现的解析器,负载均衡的逻辑在这里实现 UpdateState 核心
 │          ├─selector
 │          │  ├─node
 │          │  │  ├─direct
 │          │  │  └─ewma
 │          │  ├─p2c
 │          │  ├─random
 │          │  └─wrr
-│          └─serverinterceptors
+│          └─serverinterceptors 服务端的拦截器：超时，crash恢复
 ├─pkg
 │  ├─app 包括的项目的启动服务，配置文件的读取，命令行工具，以及其他选项
 │  │  ├─app.go 服务启动：命令行工具，日志，错误包，配置等

gmicro/registry/registry.go

@@ -1,6 +1,8 @@
 package registry
 
-import "context"
+import (
+	"context"
+)
 
 // ServiceInstance is the service instance for registry center.
 type ServiceInstance struct {

gmicro/server/rpcserver/client.go

@@ -0,0 +1,151 @@
+package rpcserver
+
+import (
+	"context"
+	"time"
+
+	"google.golang.org/grpc"
+	grpcinsecure "google.golang.org/grpc/credentials/insecure"
+
+	"github.com/CoderI421/gframework/gmicro/registry"
+	"github.com/CoderI421/gframework/gmicro/server/rpcserver/clientinterceptors"
+	"github.com/CoderI421/gframework/gmicro/server/rpcserver/resolver/discovery"
+	"github.com/CoderI421/gframework/pkg/log"
+)
+
+type ClientOption func(o *clientOptions)
+type clientOptions struct {
+	// 服务端的地址
+	endpoint string
+	// 超时时间
+	timeout time.Duration
+	// 服务发现接口
+	discovery registry.Discovery
+	// Unary 服务的拦截器
+	unaryInts []grpc.UnaryClientInterceptor
+	// Stream 服务的拦截器
+	streamInts []grpc.StreamClientInterceptor
+	// 用户自己设置 grpc 连接的结构体,例如: grpc.WithInsecure()， grpc.WithTransportCredentials()
+	rpcOpts []grpc.DialOption
+	// 根据 Name 生成负载均衡的策略
+	balancerName string
+
+	// 客户端的日志
+	logger log.Logger
+}
+
+// 设置服务端的地址
+func WithEndpoint(endpoint string) ClientOption {
+	return func(o *clientOptions) {
+		o.endpoint = endpoint
+	}
+}
+
+// 设置超时时间
+func WithClientTimeout(timeout time.Duration) ClientOption {
+	return func(o *clientOptions) {
+		o.timeout = timeout
+	}
+}
+
+// 设置服务发现
+func WithDiscovery(d registry.Discovery) ClientOption {
+	return func(o *clientOptions) {
+		o.discovery = d
+	}
+}
+
+// 设置拦截器
+func WithClientUnaryInterceptor(in ...grpc.UnaryClientInterceptor) ClientOption {
+	return func(o *clientOptions) {
+		o.unaryInts = in
+	}
+}
+
+// 设置stream拦截器
+func WithClientStreamInterceptor(in ...grpc.StreamClientInterceptor) ClientOption {
+	return func(o *clientOptions) {
+		o.streamInts = in
+	}
+}
+
+// 设置grpc的dial选项
+func WithClientOptions(opts ...grpc.DialOption) ClientOption {
+	return func(o *clientOptions) {
+		o.rpcOpts = opts
+	}
+}
+
+// 设置负载均衡器
+func WithBalancerName(name string) ClientOption {
+	return func(o *clientOptions) {
+		o.balancerName = name
+	}
+}
+
+// 设置日志
+func WithClientLogger(logger log.Logger) ClientOption {
+	return func(o *clientOptions) {
+		o.logger = logger
+	}
+}
+
+// 非安全拨号
+func DialInsecure(ctx context.Context, opts ...ClientOption) (*grpc.ClientConn, error) {
+	return dial(ctx, true, opts...)
+}
+
+func Dial(ctx context.Context, opts ...ClientOption) (*grpc.ClientConn, error) {
+	return dial(ctx, false, opts...)
+}
+
+func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.ClientConn, error) {
+	// 默认配置
+	options := clientOptions{
+		timeout:      2000 * time.Millisecond,
+		balancerName: "round_robin",
+	}
+
+	for _, o := range opts {
+		o(&options)
+	}
+
+	//TODO 客户端默认拦截器
+	ints := []grpc.UnaryClientInterceptor{
+		clientinterceptors.TimeoutInterceptor(options.timeout),
+	}
+	streamInts := []grpc.StreamClientInterceptor{}
+
+	if len(options.unaryInts) > 0 {
+		ints = append(ints, options.unaryInts...)
+	}
+	if len(options.streamInts) > 0 {
+		streamInts = append(streamInts, options.streamInts...)
+	}
+
+	//可以由用户端自己传递 这些默认的
+	grpcOpts := []grpc.DialOption{
+		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "` + options.balancerName + `"}`),
+		grpc.WithChainUnaryInterceptor(ints...),
+		grpc.WithChainStreamInterceptor(streamInts...),
+	}
+
+	//服务发现的选项
+	if &options.discovery != nil {
+		grpcOpts = append(grpcOpts, grpc.WithResolvers(
+			discovery.NewBuilder(options.discovery,
+				discovery.WithInsecure(insecure)),
+		))
+	}
+
+	// 如果是非安全模式
+	if insecure {
+		grpcOpts = append(grpcOpts, grpc.WithTransportCredentials(grpcinsecure.NewCredentials()))
+	}
+
+	if len(options.rpcOpts) > 0 {
+		grpcOpts = append(grpcOpts, options.rpcOpts...)
+	}
+
+	return grpc.DialContext(ctx, options.endpoint, grpcOpts...)
+}

gmicro/server/rpcserver/clientinterceptors/timeoutinterceptor.go

@@ -0,0 +1,22 @@
+package clientinterceptors
+
+import (
+	"context"
+	"time"
+
+	"google.golang.org/grpc"
+)
+
+func TimeoutInterceptor(timeout time.Duration) grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		if timeout <= 0 {
+			return invoker(ctx, method, req, reply, cc, opts...)
+		}
+		ctx, cancel := context.WithTimeout(ctx, timeout)
+		defer cancel()
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}
+
+//借鉴go-zero

gmicro/server/rpcserver/resolver/direct/builder.go

@@ -0,0 +1,43 @@
+package direct
+
+import (
+	"strings"
+
+	"google.golang.org/grpc/resolver"
+)
+
+type directBuilder struct{}
+
+func init() {
+	resolver.Register(NewBuilder())
+}
+
+/*
+NewBuilder creates a directBuilder which is used to factory direct resolvers.
+example:
+
+	direct://<authority>/127.0.0.1:9000
+
+到达这样一个目的
+*/
+func NewBuilder() *directBuilder {
+	return &directBuilder{}
+}
+func (d *directBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	addrs := make([]resolver.Address, 0)
+	for _, addr := range strings.Split(strings.TrimPrefix(target.URL.Path, "/"), ",") {
+		addrs = append(addrs, resolver.Address{Addr: addr})
+	}
+	//grpc建立连接的逻辑都在这里UpdateState
+	err := cc.UpdateState(resolver.State{Addresses: addrs})
+	if err != nil {
+		return nil, err
+	}
+	return newDirectResolver(), nil
+}
+
+func (d *directBuilder) Scheme() string {
+	return "direct"
+}
+
+var _ resolver.Builder = &directBuilder{}

gmicro/server/rpcserver/resolver/direct/resolver.go

@@ -0,0 +1,12 @@
+package direct
+
+import "google.golang.org/grpc/resolver"
+
+type directResolver struct{}
+
+func newDirectResolver() *directResolver {
+	return &directResolver{}
+}
+func (r *directResolver) Close() {}
+
+func (r *directResolver) ResolveNow(options resolver.ResolveNowOptions) {}

gmicro/server/rpcserver/resolver/discovery/builder.go

@@ -0,0 +1,86 @@
+package discovery
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"time"
+
+	"github.com/CoderI421/gframework/gmicro/registry"
+
+	"google.golang.org/grpc/resolver"
+)
+
+const name = "discovery"
+
+// Option is builder option.
+type Option func(o *builder)
+
+// WithTimeout with timeout option.
+func WithTimeout(timeout time.Duration) Option {
+	return func(b *builder) {
+		b.timeout = timeout
+	}
+}
+
+// WithInsecure with isSecure option.
+func WithInsecure(insecure bool) Option {
+	return func(b *builder) {
+		b.insecure = insecure
+	}
+}
+
+type builder struct {
+	discoverer registry.Discovery
+	timeout    time.Duration
+	insecure   bool
+}
+
+// NewBuilder creates a builder which is used to factory registry resolvers.
+func NewBuilder(d registry.Discovery, opts ...Option) resolver.Builder {
+	b := &builder{
+		discoverer: d,
+		timeout:    time.Second * 10,
+		insecure:   false,
+	}
+	for _, o := range opts {
+		o(b)
+	}
+	return b
+}
+
+func (b *builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	var (
+		err error
+		w   registry.Watcher
+	)
+	done := make(chan struct{}, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		w, err = b.discoverer.Watch(ctx, strings.TrimPrefix(target.URL.Path, "/"))
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(b.timeout):
+		err = errors.New("discovery create watcher overtime")
+	}
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	r := &discoveryResolver{
+		w:        w,
+		cc:       cc,
+		ctx:      ctx,
+		cancel:   cancel,
+		insecure: b.insecure,
+	}
+	go r.watch()
+	return r, nil
+}
+
+// Scheme return scheme of discovery
+func (*builder) Scheme() string {
+	return name
+}