From 0d3f26ecf17d3108921bf7a37a3e5d1cfe946b05 Mon Sep 17 00:00:00 2001 From: tangmingyou <234767776@qq.com> Date: Wed, 10 Jan 2024 00:47:38 +0800 Subject: [PATCH] grpc server --- cmd/auth/config.toml | 33 ++++++++++++++ cmd/auth/main.go | 33 +++++++++++--- cmd/gateway_ws/config.toml | 2 +- cmd/gateway_ws/main.go | 31 +++++++++---- generate.go | 6 +-- go.mod | 11 ++--- go.sum | 14 ++++-- internal/auth/logic/auth_server.go | 51 +++++++++++++++++++++- internal/gateway_ws/server/http_server.go | 5 ++- internal/gateway_ws/server/ws_server.go | 45 ++++++++++++++----- internal/gateway_ws/session/net_client.go | 9 ++-- internal/postal/logic/postal_server.go | 5 +-- pkg/grpc/discovery/instance.go | 1 - pkg/grpc/discovery/register.go | 4 +- pkg/grpc/generic/generic_client.go | 2 +- pkg/grpc/generic/generic_client_factory.go | 14 +++++- pkg/protocol/protocol.go | 22 +++++++--- pkg/protocol/protocol_test.go | 4 +- pkg/utils/resp/resp.go | 4 +- 19 files changed, 235 insertions(+), 61 deletions(-) create mode 100644 cmd/auth/config.toml diff --git a/cmd/auth/config.toml b/cmd/auth/config.toml new file mode 100644 index 0000000..9f48a60 --- /dev/null +++ b/cmd/auth/config.toml @@ -0,0 +1,33 @@ +[app] + +[grpc] +address = ":7020" +maxSendMsgSize = "8Mi" +maxRecvMsgSize = "8Mi" +readBufferSize = "8Ki" +writeBufferSize = "8Ki" + +[grpc.register.attrs] +weight = 100 + +[etcd] +endpoints = ["124.222.131.236:3279"] +username = "root" +password = "sopod@etcd" + +[redis] +Addr = "124.222.131.236:3379" +Password = "sopod@redis#" +DB = 1 +MinIdleConns = 3 + +[gorm] +logMode=true + +[gorm.mysql] +# https://gorm.io/zh_CN/docs/connecting_to_the_database.html +DSN = "root:sopod_mysql2347-@tcp(124.222.131.236:3666)/groups?charset=utf8&parseTime=True&loc=Local" + +[prometheus] +enable = false +port = 7029 diff --git a/cmd/auth/main.go b/cmd/auth/main.go index e3a82b6..9303dba 100644 --- a/cmd/auth/main.go +++ b/cmd/auth/main.go @@ -1,12 +1,35 @@ package main import ( - "encoding/binary" - "fmt" + clientv3 "go.etcd.io/etcd/client/v3" + "google.golang.org/grpc/grpclog" + "sonet/internal/auth/logic" + "sonet/pkg/config" + "sonet/pkg/grpc/discovery" + "sonet/pkg/utils/logger" + "sonet/pkg/utils/shutdown" ) func main() { - buf := make([]byte, 4) - binary.BigEndian.PutUint32(buf, 12) - fmt.Printf("%v\n", buf) + grpclog.SetLoggerV2(logger.Logger) + + conf := config.LoadConfig(nil, "cmd/auth") + + // registry and run... + etcdClient, err := clientv3.New(conf.Etcd) + if err != nil { + panic(err) + } + + registry := discovery.NewRegister(etcdClient) + shutdown.AddShutdownHook(registry.Stop) + authServer := logic.NewAuthServer() + go func() { + err = authServer.Run(conf.Grpc, registry) + if err != nil { + panic(err) + } + }() + + shutdown.Await() } diff --git a/cmd/gateway_ws/config.toml b/cmd/gateway_ws/config.toml index 9d5075d..1cca101 100644 --- a/cmd/gateway_ws/config.toml +++ b/cmd/gateway_ws/config.toml @@ -1,5 +1,5 @@ - [app] +httpPort = 7001 [grpc] address = ":7010" diff --git a/cmd/gateway_ws/main.go b/cmd/gateway_ws/main.go index fdad1c8..955fbfd 100644 --- a/cmd/gateway_ws/main.go +++ b/cmd/gateway_ws/main.go @@ -5,6 +5,8 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/grpclog" + "google.golang.org/grpc/resolver" + "sonet/internal/gateway_ws/server" "sonet/internal/postal/logic" "sonet/pkg/config" "sonet/pkg/grpc/discovery" @@ -13,11 +15,16 @@ import ( "sonet/pkg/utils/shutdown" ) +type GatewayWsConfig struct { + HttpPort int +} + // websocket server with postalService func main() { grpclog.SetLoggerV2(logger.Logger) - conf := config.LoadConfig(nil, "cmd/gateway_ws") + appConf := &GatewayWsConfig{} + conf := config.LoadConfig(appConf, "cmd/gateway_ws") // registry and run... etcdClient, err := clientv3.New(conf.Etcd) @@ -25,17 +32,25 @@ func main() { panic(err) } - resolver := discovery.NewResolver(etcdClient) + etcdResolver := discovery.NewResolver(etcdClient) + resolver.Register(etcdResolver) - generic.NewGpcGenericClientFactory(resolver, grpc.WithTransportCredentials(insecure.NewCredentials())) - //server.NewConnHandler() - //server.NewHttpServer() + grpcFactory := generic.NewGpcGenericClientFactory(etcdResolver, grpc.WithTransportCredentials(insecure.NewCredentials())) + grpcFactory.Init() + connHandler := server.NewConnHandler(grpcFactory) + httpServer := server.NewHttpServer(connHandler) + go func() { + err = httpServer.Run(appConf.HttpPort) + if err != nil { + panic(err) + } + }() - postalRegistry := discovery.NewRegister(etcdClient) - shutdown.AddShutdownHook(postalRegistry.Stop) + registry := discovery.NewRegister(etcdClient) + shutdown.AddShutdownHook(registry.Stop) postalServer := logic.NewPostalServer() go func() { - err = postalServer.Run(conf.Grpc, postalRegistry) + err = postalServer.Run(conf.Grpc, registry) if err != nil { panic(err) } diff --git a/generate.go b/generate.go index 6fa3169..b6e80e3 100644 --- a/generate.go +++ b/generate.go @@ -14,12 +14,10 @@ func main() { } func beforeGenerate() { - err := os.MkdirAll("api/gen", os.ModePerm) - if err != nil { + if err := os.MkdirAll("api/gen", os.ModePerm); err != nil { panic(err) } - err = os.MkdirAll("api/genjs", os.ModePerm) - if err != nil { + if err := os.MkdirAll("api/genjs", os.ModePerm); err != nil { panic(err) } } diff --git a/go.mod b/go.mod index d6d100f..b071139 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,11 @@ module sonet go 1.19 require ( + github.com/bytedance/sonic v1.10.2 + github.com/dsnet/golib/unitconv v1.0.2 + github.com/gin-gonic/gin v1.9.1 github.com/golang/protobuf v1.5.3 + github.com/gorilla/websocket v1.5.1 github.com/jhump/protoreflect v1.15.4 github.com/nats-io/nats.go v1.31.0 github.com/redis/go-redis/v9 v9.4.0 @@ -18,24 +22,21 @@ require ( require ( github.com/bufbuild/protocompile v0.7.1 // indirect - github.com/bytedance/sonic v1.9.1 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect - github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect + github.com/chenzhuoyu/base64x v0.0.0-20230717121745-296ad89f973d // indirect + github.com/chenzhuoyu/iasm v0.9.1 // indirect github.com/coreos/go-semver v0.3.0 // indirect github.com/coreos/go-systemd/v22 v22.3.2 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect - github.com/dsnet/golib/unitconv v1.0.2 // indirect github.com/fsnotify/fsnotify v1.7.0 // indirect github.com/gabriel-vasile/mimetype v1.4.2 // indirect github.com/gin-contrib/sse v0.1.0 // indirect - github.com/gin-gonic/gin v1.9.1 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.14.0 // indirect github.com/go-sql-driver/mysql v1.7.0 // indirect github.com/goccy/go-json v0.10.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect - github.com/gorilla/websocket v1.5.1 // indirect github.com/hashicorp/hcl v1.0.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect diff --git a/go.sum b/go.sum index 016feda..f139422 100644 --- a/go.sum +++ b/go.sum @@ -5,13 +5,18 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bufbuild/protocompile v0.7.1 h1:Kd8fb6EshOHXNNRtYAmLAwy/PotlyFoN0iMbuwGNh0M= github.com/bufbuild/protocompile v0.7.1/go.mod h1:+Etjg4guZoAqzVk2czwEQP12yaxLJ8DxuqCJ9qHdH94= github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM= -github.com/bytedance/sonic v1.9.1 h1:6iJ6NqdoxCDr6mbY8h18oSO+cShGSMRGCEo7F2h0x8s= -github.com/bytedance/sonic v1.9.1/go.mod h1:i736AoUSYt75HyZLoJW9ERYxcy6eaN6h4BZXU064P/U= +github.com/bytedance/sonic v1.10.0-rc/go.mod h1:ElCzW+ufi8qKqNW0FY314xriJhyJhuoJ3gFZdAHF7NM= +github.com/bytedance/sonic v1.10.2 h1:GQebETVBxYB7JGWJtLBi07OVzWwt+8dWA00gEVW2ZFE= +github.com/bytedance/sonic v1.10.2/go.mod h1:iZcSUejdk5aukTND/Eu/ivjQuEL0Cu9/rf50Hi0u/g4= github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/chenzhuoyu/base64x v0.0.0-20211019084208-fb5309c8db06/go.mod h1:DH46F32mSOjUmXrMHnKwZdA8wcEefY7UVqBKYGjpdQY= -github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 h1:qSGYFH7+jGhDF8vLC+iwCD4WpbV1EBDSzWkJODFLams= github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311/go.mod h1:b583jCggY9gE99b6G5LEC39OIiVsWj+R97kbl5odCEk= +github.com/chenzhuoyu/base64x v0.0.0-20230717121745-296ad89f973d h1:77cEq6EriyTZ0g/qfRdp61a3Uu/AWrgIq2s0ClJV1g0= +github.com/chenzhuoyu/base64x v0.0.0-20230717121745-296ad89f973d/go.mod h1:8EPpVsBuRksnlj1mLy4AWzRNQYxauNi62uWcE3to6eA= +github.com/chenzhuoyu/iasm v0.9.0/go.mod h1:Xjy2NpN3h7aUqeqM+woSuuvxmIe6+DDsiNLIrkAmYog= +github.com/chenzhuoyu/iasm v0.9.1 h1:tUHQJXo3NhBqw6s33wkGn9SP3bvrWLdlVIJ3hQBL7P0= +github.com/chenzhuoyu/iasm v0.9.1/go.mod h1:Xjy2NpN3h7aUqeqM+woSuuvxmIe6+DDsiNLIrkAmYog= github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM= github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI= @@ -32,6 +37,7 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg= github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU= +github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY= github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY= @@ -71,6 +77,7 @@ github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQs github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk= github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY= +github.com/knz/go-libedit v1.10.1/go.mod h1:MZTVkCWyz0oBc7JOWP3wNAzd002ZbM/5hgShxwh4x8M= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= @@ -238,4 +245,5 @@ gorm.io/driver/mysql v1.5.2/go.mod h1:pQLhh1Ut/WUAySdTHwBpBv6+JKcj+ua4ZFx1QQTBzb gorm.io/gorm v1.25.2-0.20230530020048-26663ab9bf55/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k= gorm.io/gorm v1.25.5 h1:zR9lOiiYf09VNh5Q1gphfyia1JpiClIWG9hQaxB/mls= gorm.io/gorm v1.25.5/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= +nullprogram.com/x/optparse v1.0.0/go.mod h1:KdyPE+Igbe0jQUrVfMqDMeJQIJZEuyV7pjYmp6pbG50= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/internal/auth/logic/auth_server.go b/internal/auth/logic/auth_server.go index 205ec15..7432e93 100644 --- a/internal/auth/logic/auth_server.go +++ b/internal/auth/logic/auth_server.go @@ -2,22 +2,71 @@ package logic import ( "context" + "errors" + "google.golang.org/grpc" + "google.golang.org/grpc/reflection" + "net" "sonet/api/gen/auth" + "sonet/pkg/config" + "sonet/pkg/grpc/discovery" + "sonet/pkg/utils/logger" ) type AuthServer struct { auth.UnimplementedAuthServer } +func NewAuthServer() *AuthServer { + return &AuthServer{} +} + +func (s *AuthServer) Run(conf config.GrpcConfig, register *discovery.Register) (err error) { + server := grpc.NewServer( + config.GetGrpcOptions(conf)..., + ) + if !conf.NoReflection { + // 注册反射服务 + reflection.Register(server) + } + auth.RegisterAuthServer(server, s) + listen, err := net.Listen("tcp", conf.Address) + if err != nil { + return + } + + // registry discovery + registerConf := conf.Register + if registerConf.Name == "" { + registerConf.Name = auth.Auth_ServiceDesc.ServiceName + } + if registerConf.Addr == "" { + registerConf.Addr, err = discovery.RegisterAddress(listen) + if err != nil { + return + } + } + if err = register.Register(registerConf); err != nil { + return + } + + // run serve + logger.Infof("%s grpc server running %s\n", registerConf.Name, listen.Addr().String()) + err = server.Serve(listen) + return +} + func (s *AuthServer) Login(ctx context.Context, req *auth.ReqLogin) (*auth.ResLogin, error) { return nil, nil } + func (s *AuthServer) Verify(ctx context.Context, req *auth.ReqVerify) (*auth.Subject, error) { - return nil, nil + return &auth.Subject{Username: "sunshine"}, errors.New("test error...") } + func (s *AuthServer) Registry(ctx context.Context, req *auth.ReqRegistry) (*auth.Subject, error) { return nil, nil } + func (s *AuthServer) FindByUid(ctx context.Context, req *auth.ReqFindByUid) (*auth.Subject, error) { return nil, nil } diff --git a/internal/gateway_ws/server/http_server.go b/internal/gateway_ws/server/http_server.go index 666613e..e93224e 100644 --- a/internal/gateway_ws/server/http_server.go +++ b/internal/gateway_ws/server/http_server.go @@ -6,6 +6,7 @@ import ( "github.com/gorilla/websocket" "net/http" "sonet/pkg/utils/conver" + "sonet/pkg/utils/logger" "sonet/pkg/utils/resp" "time" ) @@ -23,7 +24,9 @@ func NewHttpServer(connHandler *ConnHandler) *HttpServer { func (s *HttpServer) Run(port int) error { server := gin.Default() server.GET("/ws", s.upgrade) - return server.Run(fmt.Sprintf(":%d", port)) + addr := fmt.Sprintf(":%d", port) + logger.Infof("http server run %s", addr) + return server.Run(addr) } var ( diff --git a/internal/gateway_ws/server/ws_server.go b/internal/gateway_ws/server/ws_server.go index e07d1d6..794401c 100644 --- a/internal/gateway_ws/server/ws_server.go +++ b/internal/gateway_ws/server/ws_server.go @@ -2,7 +2,6 @@ package server import ( "context" - "fmt" "github.com/gorilla/websocket" "runtime/debug" "sonet/internal/gateway_ws/session" @@ -25,6 +24,8 @@ func (c *ConnHandler) handleConn(conn *websocket.Conn) { client := session.NewNetClient(conn, ReadDeadline, WriteDeadline) defer func() { + // TODO 连接关闭，mq发送关闭事件 + // 捕获其他错误 if r := recover(); r != nil { logger.Error("NetClient recover error: ", r) @@ -36,9 +37,8 @@ func (c *ConnHandler) handleConn(conn *websocket.Conn) { defer client.Close() for { - message, err := client.ReadMessage() + ignore, message, err := client.ReadMessage() if err != nil { - // TODO 连接关闭，mq发送关闭事件 if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) { logger.Error("unexpected close error: ", err) } else { @@ -47,38 +47,41 @@ func (c *ConnHandler) handleConn(conn *websocket.Conn) { } return } + if ignore { + continue + } - payload, err := protocol.Decode(message) + payload, err := protocol.DecodeSo(message) if err != nil { logger.Errorf("decode message error: len=%d", len(message), err) return } - fmt.Printf("%v\n", payload) + header := payload.Header // grpc generic call ctx := context.Background() - grpcClient, err := c.grpcFactory.GetClient(ctx, payload.Header.Svc) + grpcClient, err := c.grpcFactory.GetClient(ctx, header.Svc) if err != nil { logger.Error("get grpc generic client error: ", err) continue } - resp, err := grpcClient.InvokeUnary(ctx, payload.Header.Target, payload.Body) + resp, err := grpcClient.InvokeUnary(ctx, header.Target, payload.Body) if err != nil { + // todo 提取 err: fmt.Sprintf("rpc error: code = %s desc = %s", s.Code(), s.Message()) + writeError(client, header.SeqId, err.Error()) logger.Error("grpc generic call error: ", err) continue } - fmt.Println(resp) // write response - payload.Header.Type = protocol.TypeResponse - payload.Header.Svc = "" - payload.Header.Target = "" + header.Type = protocol.TypeResponse + header.Target = resp.XXX_MessageName() payload.Body, err = resp.Marshal() if err != nil { logger.Error("generic call response marshal error: ", err) continue } - resMessage, err := protocol.Encode(payload) + resMessage, err := protocol.EncodeSo(payload) if err != nil { logger.Error("grpc generic call error: ", err) continue @@ -87,3 +90,21 @@ func (c *ConnHandler) handleConn(conn *websocket.Conn) { } } + +func writeError(client *session.NetClient, seqId int32, errMsg string) { + header := &protocol.Header{} + payload := &protocol.Payload{Header: header} + + header.Magic = protocol.Magic + header.Type = protocol.TypeError + header.Status = 50 + header.SeqId = seqId + payload.Body = []byte(errMsg) + message, err := protocol.EncodeSo(payload) + if err != nil { + logger.Error("NetClient writeError encode error:", err) + return + } + client.MustWrite(message) + return +} diff --git a/internal/gateway_ws/session/net_client.go b/internal/gateway_ws/session/net_client.go index 647ac61..301ec49 100644 --- a/internal/gateway_ws/session/net_client.go +++ b/internal/gateway_ws/session/net_client.go @@ -61,8 +61,10 @@ func (c *NetClient) MustWrite(bytes []byte) { } } -func (c *NetClient) ReadMessage() (bytes []byte, err error) { - err = c.Conn.SetReadDeadline(time.Now().Add(c.readDeadline)) +// ReadMessage 阻塞读取消息 +func (c *NetClient) ReadMessage() (ignore bool, bytes []byte, err error) { + // err = c.Conn.SetReadDeadline(time.Now().Add(c.readDeadline)) + err = c.Conn.SetReadDeadline(time.Time{}) if err != nil { return } @@ -70,7 +72,8 @@ func (c *NetClient) ReadMessage() (bytes []byte, err error) { var messageType int messageType, bytes, err = c.Conn.ReadMessage() if messageType != websocket.BinaryMessage { - err = fmt.Errorf("only support websocket binary message") + ignore = true + logger.Infof("read websocket message type: %d", messageType) return } return diff --git a/internal/postal/logic/postal_server.go b/internal/postal/logic/postal_server.go index 0bedd4b..5f87e55 100644 --- a/internal/postal/logic/postal_server.go +++ b/internal/postal/logic/postal_server.go @@ -45,10 +45,7 @@ func (p *PostalServer) Run(conf config.GrpcConfig, postalRegister *discovery.Reg return } } - if regConf.Ttl == 0 { - regConf.Ttl = discovery.DefaultRegisterTTL - } - if err = postalRegister.Register(regConf, regConf.Ttl); err != nil { + if err = postalRegister.Register(regConf); err != nil { return } diff --git a/pkg/grpc/discovery/instance.go b/pkg/grpc/discovery/instance.go index 8d0c93b..2409ba6 100644 --- a/pkg/grpc/discovery/instance.go +++ b/pkg/grpc/discovery/instance.go @@ -13,7 +13,6 @@ type Server struct { Name string `json:"name"` Addr string `json:"addr"` // 地址 Attrs map[string]string `json:"attrs"` // attributes - Ttl int64 `json:"-"` } func BuildPrefix(server Server) string { diff --git a/pkg/grpc/discovery/register.go b/pkg/grpc/discovery/register.go index 1402dd4..531337d 100644 --- a/pkg/grpc/discovery/register.go +++ b/pkg/grpc/discovery/register.go @@ -56,13 +56,13 @@ func NewRegister(client *clientv3.Client) *Register { } // Register a user -func (r *Register) Register(srvInfo Server, ttl int64) (err error) { +func (r *Register) Register(srvInfo Server) (err error) { if strings.Split(srvInfo.Addr, ":")[0] == "" { return errors.New("invalid ip address") } r.srvInfo = srvInfo - r.srvTTL = ttl + r.srvTTL = DefaultRegisterTTL if err = r.register(); err != nil { return err diff --git a/pkg/grpc/generic/generic_client.go b/pkg/grpc/generic/generic_client.go index 0bf2cf1..2875290 100644 --- a/pkg/grpc/generic/generic_client.go +++ b/pkg/grpc/generic/generic_client.go @@ -33,7 +33,7 @@ func NewGpcGenericClient(serviceName string, conn *grpc.ClientConn) *GrpcGeneric func (c *GrpcGenericClient) Init(ctx context.Context) (err error) { // fetch service description refClient := grpcreflect.NewClientV1Alpha(ctx, grpc_reflection_v1alpha.NewServerReflectionClient(c.conn)) - desc_source.DescriptorSourceFromServer(ctx, refClient) + c.descSource = desc_source.DescriptorSourceFromServer(ctx, refClient) dsc, e1 := c.descSource.FindSymbol(c.serviceName) if e1 != nil { err = fmt.Errorf("service %s not found", c.serviceName) diff --git a/pkg/grpc/generic/generic_client_factory.go b/pkg/grpc/generic/generic_client_factory.go index 6d48291..5caf4d3 100644 --- a/pkg/grpc/generic/generic_client_factory.go +++ b/pkg/grpc/generic/generic_client_factory.go @@ -2,8 +2,10 @@ package generic import ( "context" + "errors" "fmt" "google.golang.org/grpc" + "runtime/debug" "sonet/pkg/grpc/discovery" "sync" ) @@ -26,11 +28,21 @@ func (f *GrpcGenericClientFactory) Init() { } func (f *GrpcGenericClientFactory) NewClient(ctx context.Context, serviceName string, opts ...grpc.DialOption) (client *GrpcGenericClient, err error) { + defer func() { + if r := recover(); r != nil { + fmt.Println(r) + fmt.Println(string(debug.Stack())) + err = errors.New("errrrr") + } + }() addr := fmt.Sprintf("%s:///%s", f.resolver.Scheme(), serviceName) - dialOpts := make([]grpc.DialOption, len(f.defaultOpts)+len(opts)) + dialOpts := make([]grpc.DialOption, 0, len(f.defaultOpts)+len(opts)) dialOpts = append(dialOpts, f.defaultOpts...) dialOpts = append(dialOpts, opts...) conn, err := grpc.DialContext(ctx, addr, dialOpts...) + if err != nil { + return + } client = NewGpcGenericClient(serviceName, conn) err = client.Init(ctx) return diff --git a/pkg/protocol/protocol.go b/pkg/protocol/protocol.go index 596c326..13fdf1f 100644 --- a/pkg/protocol/protocol.go +++ b/pkg/protocol/protocol.go @@ -38,12 +38,12 @@ type Header struct { } type Payload struct { - Header Header + Header *Header Body []byte } -func Decode(bytes []byte) (payload *Payload, err error) { - header := Header{} +func DecodeSo(bytes []byte) (payload *Payload, err error) { + header := &Header{} header.Magic = bytes[0] if header.Magic != Magic { @@ -55,6 +55,11 @@ func Decode(bytes []byte) (payload *Payload, err error) { header.UrlType = bytes[3] >> 4 header.SerializeType = bytes[3] & 0xF header.SeqId = int32(binary.BigEndian.Uint32(bytes[4:8])) + // error message + if header.Type == TypeError { + payload.Body = bytes[8:] + return + } var cursor int switch header.UrlType { @@ -80,12 +85,13 @@ func Decode(bytes []byte) (payload *Payload, err error) { return } -func Encode(payload *Payload) (bytes []byte, err error) { +func EncodeSo(payload *Payload) (bytes []byte, err error) { header := payload.Header headerLen := 16 var svc, target []byte var svcLen, targetLen, bodyLen int - if header.UrlType == 1 { + // 定位符为字符串 + if header.Type != TypeError && header.UrlType == 1 { svc = []byte(header.Svc) target = []byte(header.Target) svcLen = len(svc) @@ -99,6 +105,12 @@ func Encode(payload *Payload) (bytes []byte, err error) { bytes[2] = header.Status bytes[3] = (header.UrlType << 4) | header.SerializeType binary.BigEndian.PutUint32(bytes[4:8], uint32(header.SeqId)) + if header.Type == TypeError { + if bodyLen > 0 { + copy(bytes[8:], payload.Body) + } + return + } var cursor int switch header.UrlType { diff --git a/pkg/protocol/protocol_test.go b/pkg/protocol/protocol_test.go index f1193d7..8619b0f 100644 --- a/pkg/protocol/protocol_test.go +++ b/pkg/protocol/protocol_test.go @@ -20,12 +20,12 @@ func TestProtocolCodec(t *testing.T) { Header: header, Body: []byte(`{"receiver":"10001啊"}`), } - bytes, err := Encode(payload) + bytes, err := EncodeSo(payload) if err != nil { t.Error(err) return } - payload2, err := Decode(bytes) + payload2, err := DecodeSo(bytes) if err != nil { t.Error(err) return diff --git a/pkg/utils/resp/resp.go b/pkg/utils/resp/resp.go index 098e42a..e12478c 100644 --- a/pkg/utils/resp/resp.go +++ b/pkg/utils/resp/resp.go @@ -2,7 +2,7 @@ package resp import ( "github.com/bytedance/sonic" - "github.com/cloudwego/kitex/pkg/klog" + "sonet/pkg/utils/logger" ) const ( @@ -25,7 +25,7 @@ type Response struct { func (resp *Response) Json() []byte { json, err := sonic.Marshal(resp) if err != nil { - klog.Error("unknown json error: ", err) + logger.Error("unknown json error: ", err) return nil } return json