Merge pull request 'check subscribe errors' (#164) from subscribeerr into v3

Reviewed-on: #164

.github/workflows/autoapprove.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: approve
-        uses: hmarr/auto-approve-action@v2
+        uses: hmarr/auto-approve-action@v3
         if: github.actor == 'vtolstov' || github.actor == 'dependabot[bot]'
         id: approve
         with:

.github/workflows/build.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
     - name: checkout
@@ -34,7 +34,7 @@ jobs:
       - name: checkout
         uses: actions/checkout@v3
       - name: lint
-        uses: golangci/golangci-lint-action@v3.1.0
+        uses: golangci/golangci-lint-action@v3.4.0
         continue-on-error: true
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.

.github/workflows/codeql-analysis.yml

@@ -45,12 +45,12 @@ jobs:
     - name: checkout
       uses: actions/checkout@v3
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
     # Initializes the CodeQL tools for scanning.
     - name: init
-      uses: github/codeql-action/init@v1
+      uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -61,7 +61,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: autobuild
-      uses: github/codeql-action/autobuild@v1
+      uses: github/codeql-action/autobuild@v2
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -75,4 +75,4 @@ jobs:
     #   make release
 
     - name: analyze
-      uses: github/codeql-action/analyze@v1
+      uses: github/codeql-action/analyze@v2

.github/workflows/dependabot-automerge.yml

@@ -15,7 +15,7 @@ jobs:
     steps:
       - name: metadata
         id: metadata
-        uses: dependabot/fetch-metadata@v1.3.0
+        uses: dependabot/fetch-metadata@v1.3.6
         with:
           github-token: "${{ secrets.TOKEN }}"
       - name: merge

.github/workflows/pr.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.17
     - name: checkout
@@ -34,7 +34,7 @@ jobs:
       - name: checkout
         uses: actions/checkout@v3
       - name: lint
-        uses: golangci/golangci-lint-action@v3.1.0
+        uses: golangci/golangci-lint-action@v3.4.0
         continue-on-error: true
         with:
           # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.

go.mod

@@ -4,8 +4,8 @@ go 1.16
 
 require (
 	github.com/golang/protobuf v1.5.2
-	go.unistack.org/micro/v3 v3.9.1
-	golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b
-	google.golang.org/grpc v1.45.0
-	google.golang.org/protobuf v1.28.0
+	go.unistack.org/micro/v3 v3.10.14
+	golang.org/x/net v0.5.0
+	google.golang.org/grpc v1.52.3
+	google.golang.org/protobuf v1.28.1
 )

grpc.go

@@ -35,17 +35,17 @@ import (
 )
 
 const (
-	DefaultContentType = "application/grpc+proto"
+	DefaultContentType = "application/grpc"
 )
 
 /*
-type grpcServerReflection struct {
+type ServerReflection struct {
 	srv *grpc.Server
 	s   *serverReflectionServer
 }
 */
 
-type grpcServer struct {
+type Server struct {
 	handlers    map[string]server.Handler
 	srv         *grpc.Server
 	exit        chan chan error
@@ -55,15 +55,14 @@ type grpcServer struct {
 	rpc         *rServer
 	opts        server.Options
 	sync.RWMutex
-	init       bool
 	started    bool
 	registered bool
 	reflection bool
 }
 
-func newGRPCServer(opts ...server.Option) server.Server {
+func newServer(opts ...server.Option) *Server {
 	// create a grpc server
-	g := &grpcServer{
+	g := &Server{
 		opts: server.NewOptions(opts...),
 		rpc: &rServer{
 			serviceMap: make(map[string]*service),
@@ -92,7 +91,7 @@ func (r grpcRouter) ServeRequest(ctx context.Context, req server.Request, rsp se
 
 */
 
-func (g *grpcServer) configure(opts ...server.Option) error {
+func (g *Server) configure(opts ...server.Option) error {
 	g.Lock()
 	defer g.Unlock()
 
@@ -109,9 +108,6 @@ func (g *grpcServer) configure(opts ...server.Option) error {
 	if err := g.opts.Tracer.Init(); err != nil {
 		return err
 	}
-	if err := g.opts.Auth.Init(); err != nil {
-		return err
-	}
 	if err := g.opts.Logger.Init(); err != nil {
 		return err
 	}
@@ -132,6 +128,10 @@ func (g *grpcServer) configure(opts ...server.Option) error {
 		}
 	}
 
+	for _, k := range g.opts.Codecs {
+		encoding.RegisterCodec(&wrapMicroCodec{k})
+	}
+
 	maxMsgSize := g.getMaxMsgSize()
 
 	gopts := []grpc.ServerOption{
@@ -166,12 +166,10 @@ func (g *grpcServer) configure(opts ...server.Option) error {
 		return g.Start()
 	}
 
-	g.init = true
-
 	return nil
 }
 
-func (g *grpcServer) getMaxMsgSize() int {
+func (g *Server) getMaxMsgSize() int {
 	if g.opts.Context == nil {
 		return codec.DefaultMaxMsgSize
 	}
@@ -182,14 +180,14 @@ func (g *grpcServer) getMaxMsgSize() int {
 	return s
 }
 
-func (g *grpcServer) getCredentials() credentials.TransportCredentials {
+func (g *Server) getCredentials() credentials.TransportCredentials {
 	if g.opts.TLSConfig != nil {
 		return credentials.NewTLS(g.opts.TLSConfig)
 	}
 	return nil
 }
 
-func (g *grpcServer) getGrpcOptions() []grpc.ServerOption {
+func (g *Server) getGrpcOptions() []grpc.ServerOption {
 	if g.opts.Context == nil {
 		return nil
 	}
@@ -202,32 +200,7 @@ func (g *grpcServer) getGrpcOptions() []grpc.ServerOption {
 	return opts
 }
 
-func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err error) {
-	defer func() {
-		if r := recover(); r != nil {
-			g.RLock()
-			config := g.opts
-			g.RUnlock()
-			if config.Logger.V(logger.ErrorLevel) {
-				config.Logger.Error(config.Context, "panic recovered: ", r)
-				config.Logger.Error(config.Context, string(debug.Stack()))
-			}
-			err = errors.InternalServerError(g.opts.Name, "panic recovered: %v", r)
-		} else if err != nil {
-			g.RLock()
-			config := g.opts
-			g.RUnlock()
-			if config.Logger.V(logger.ErrorLevel) {
-				config.Logger.Errorf(config.Context, "grpc handler got error: %s", err)
-			}
-		}
-	}()
-
-	if g.wg != nil {
-		g.wg.Add(1)
-		defer g.wg.Done()
-	}
-
+func (g *Server) handler(srv interface{}, stream grpc.ServerStream) (err error) {
 	fullMethod, ok := grpc.MethodFromServerStream(stream)
 	if !ok {
 		return status.Errorf(codes.Internal, "method does not exist in context")
@@ -238,6 +211,31 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
 		return status.New(codes.InvalidArgument, err.Error()).Err()
 	}
 
+	defer func() {
+		if r := recover(); r != nil {
+			g.RLock()
+			config := g.opts
+			g.RUnlock()
+			if config.Logger.V(logger.ErrorLevel) {
+				config.Logger.Errorf(config.Context, "panic in %s.%s recovered: %v", serviceName, methodName, r)
+				config.Logger.Error(config.Context, string(debug.Stack()))
+			}
+			err = errors.InternalServerError(g.opts.Name, "panic in %s.%s recovered: %v", serviceName, methodName, r)
+		} else if err != nil {
+			g.RLock()
+			config := g.opts
+			g.RUnlock()
+			if config.Logger.V(logger.ErrorLevel) {
+				config.Logger.Errorf(config.Context, "grpc handler %s.%s got error: %s", serviceName, methodName, err)
+			}
+		}
+	}()
+
+	if g.wg != nil {
+		g.wg.Add(1)
+		defer g.wg.Done()
+	}
+
 	// get grpc metadata
 	gmd, ok := gmetadata.FromIncomingContext(stream.Context())
 	if !ok {
@@ -258,7 +256,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
 	if v, ok := md.Get("Grpc-Timeout"); ok {
 		md.Del("Grpc-Timeout")
 		td = v[:len(v)-1]
-		switch v[:] {
+		switch v[len(v)-1:] {
 		case "S":
 			td += "s"
 		case "M":
@@ -289,7 +287,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
 
 	// get peer from context
 	if p, ok := peer.FromContext(stream.Context()); ok {
-		md["Remote"] = p.Addr.String()
+		md.Set("Remote", p.Addr.String())
 		ctx = peer.NewContext(ctx, p)
 	}
 
@@ -308,7 +306,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
 
 	/*
 		if svc == nil && g.reflection && methodName == "ServerReflectionInfo" {
-			rfl := &grpcServerReflection{srv: g.srv, s: &serverReflectionServer{s: g.srv}}
+			rfl := &ServerReflection{srv: g.srv, s: &serverReflectionServer{s: g.srv}}
 			svc = &service{}
 			svc.typ = reflect.TypeOf(rfl)
 			svc.rcvr = reflect.ValueOf(rfl)
@@ -356,7 +354,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
 	return g.processStream(ctx, stream, svc, mtype, ct)
 }
 
-func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStream, service *service, mtype *methodType, ct string) error {
+func (g *Server) processRequest(ctx context.Context, stream grpc.ServerStream, service *service, mtype *methodType, ct string) error {
 	// for {
 	var err error
 	var argv, replyv reflect.Value
@@ -371,8 +369,7 @@ func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStrea
 	}
 
 	// Unmarshal request
-	// TODO: avoid Marshal call later by recv to frame and reuse it data
-	if err := stream.RecvMsg(argv.Interface()); err != nil {
+	if err = stream.RecvMsg(argv.Interface()); err != nil {
 		return err
 	}
 
@@ -391,11 +388,12 @@ func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStrea
 		service:     g.opts.Name,
 		contentType: ct,
 		method:      fmt.Sprintf("%s.%s", service.name, mtype.method.Name),
+		endpoint:    fmt.Sprintf("%s.%s", service.name, mtype.method.Name),
 		payload:     argv.Interface(),
 	}
 	// define the handler func
 	fn := func(ctx context.Context, req server.Request, rsp interface{}) (err error) {
-		returnValues = function.Call([]reflect.Value{service.rcvr, mtype.prepareContext(ctx), reflect.ValueOf(argv.Interface()), reflect.ValueOf(rsp)})
+		returnValues = function.Call([]reflect.Value{service.rcvr, mtype.prepareContext(ctx), argv, reflect.ValueOf(rsp)})
 
 		// The return value for the method is an error.
 		if rerr := returnValues[0].Interface(); rerr != nil {
@@ -413,7 +411,13 @@ func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStrea
 	statusCode := codes.OK
 	statusDesc := ""
 	// execute the handler
-	if appErr := fn(ctx, r, replyv.Interface()); appErr != nil {
+	appErr := fn(ctx, r, replyv.Interface())
+	if outmd, ok := metadata.FromOutgoingContext(ctx); ok {
+		if err = stream.SendHeader(gmetadata.New(outmd)); err != nil {
+			return err
+		}
+	}
+	if appErr != nil {
 		var errStatus *status.Status
 		switch verr := appErr.(type) {
 		case *errors.Error:
@@ -428,6 +432,8 @@ func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStrea
 			if err != nil {
 				return err
 			}
+		case (interface{ GRPCStatus() *status.Status }):
+			errStatus = verr.GRPCStatus()
 		default:
 			g.RLock()
 			config := g.opts
@@ -491,18 +497,19 @@ func (s *reflectStream) RecvMsg(m interface{}) error {
 	return s.stream.Recv(m)
 }
 
-func (g *grpcServerReflection) ServerReflectionInfo(ctx context.Context, stream server.Stream) error {
+func (g *ServerReflection) ServerReflectionInfo(ctx context.Context, stream server.Stream) error {
 	return g.s.ServerReflectionInfo(&reflectStream{stream})
 }
 */
 
-func (g *grpcServer) processStream(ctx context.Context, stream grpc.ServerStream, service *service, mtype *methodType, ct string) error {
+func (g *Server) processStream(ctx context.Context, stream grpc.ServerStream, service *service, mtype *methodType, ct string) error {
 	opts := g.opts
 
 	r := &rpcRequest{
 		service:     opts.Name,
 		contentType: ct,
 		method:      fmt.Sprintf("%s.%s", service.name, mtype.method.Name),
+		endpoint:    fmt.Sprintf("%s.%s", service.name, mtype.method.Name),
 		stream:      true,
 	}
 
@@ -531,7 +538,13 @@ func (g *grpcServer) processStream(ctx context.Context, stream grpc.ServerStream
 	statusCode := codes.OK
 	statusDesc := ""
 
-	if appErr := fn(ctx, r, ss); appErr != nil {
+	appErr := fn(ctx, r, ss)
+	if outmd, ok := metadata.FromOutgoingContext(ctx); ok {
+		if err := stream.SendHeader(gmetadata.New(outmd)); err != nil {
+			return err
+		}
+	}
+	if appErr != nil {
 		var err error
 		var errStatus *status.Status
 		switch verr := appErr.(type) {
@@ -563,7 +576,7 @@ func (g *grpcServer) processStream(ctx context.Context, stream grpc.ServerStream
 	return status.New(statusCode, statusDesc).Err()
 }
 
-func (g *grpcServer) newCodec(ct string) (codec.Codec, error) {
+func (g *Server) newCodec(ct string) (codec.Codec, error) {
 	g.RLock()
 	defer g.RUnlock()
 
@@ -578,7 +591,7 @@ func (g *grpcServer) newCodec(ct string) (codec.Codec, error) {
 	return nil, codec.ErrUnknownContentType
 }
 
-func (g *grpcServer) Options() server.Options {
+func (g *Server) Options() server.Options {
 	g.RLock()
 	opts := g.opts
 	g.RUnlock()
@@ -586,18 +599,15 @@ func (g *grpcServer) Options() server.Options {
 	return opts
 }
 
-func (g *grpcServer) Init(opts ...server.Option) error {
-	if len(opts) == 0 && g.init {
-		return nil
-	}
+func (g *Server) Init(opts ...server.Option) error {
 	return g.configure(opts...)
 }
 
-func (g *grpcServer) NewHandler(h interface{}, opts ...server.HandlerOption) server.Handler {
+func (g *Server) NewHandler(h interface{}, opts ...server.HandlerOption) server.Handler {
 	return newRPCHandler(h, opts...)
 }
 
-func (g *grpcServer) Handle(h server.Handler) error {
+func (g *Server) Handle(h server.Handler) error {
 	if err := g.rpc.register(h.Handler()); err != nil {
 		return err
 	}
@@ -606,11 +616,11 @@ func (g *grpcServer) Handle(h server.Handler) error {
 	return nil
 }
 
-func (g *grpcServer) NewSubscriber(topic string, sb interface{}, opts ...server.SubscriberOption) server.Subscriber {
+func (g *Server) NewSubscriber(topic string, sb interface{}, opts ...server.SubscriberOption) server.Subscriber {
 	return newSubscriber(topic, sb, opts...)
 }
 
-func (g *grpcServer) Subscribe(sb server.Subscriber) error {
+func (g *Server) Subscribe(sb server.Subscriber) error {
 	sub, ok := sb.(*subscriber)
 	if !ok {
 		return fmt.Errorf("invalid subscriber: expected *subscriber")
@@ -634,7 +644,7 @@ func (g *grpcServer) Subscribe(sb server.Subscriber) error {
 	return nil
 }
 
-func (g *grpcServer) Register() error {
+func (g *Server) Register() error {
 	g.RLock()
 	rsvc := g.rsvc
 	config := g.opts
@@ -708,37 +718,13 @@ func (g *grpcServer) Register() error {
 	g.Lock()
 	defer g.Unlock()
 
-	for sb := range g.subscribers {
-		handler := g.createSubHandler(sb, config)
-		var opts []broker.SubscribeOption
-		if queue := sb.Options().Queue; len(queue) > 0 {
-			opts = append(opts, broker.SubscribeGroup(queue))
-		}
-
-		subCtx := config.Context
-		if cx := sb.Options().Context; cx != nil {
-			subCtx = cx
-		}
-		opts = append(opts, broker.SubscribeContext(subCtx))
-		opts = append(opts, broker.SubscribeAutoAck(sb.Options().AutoAck))
-
-		if config.Logger.V(logger.InfoLevel) {
-			config.Logger.Infof(config.Context, "Subscribing to topic: %s", sb.Topic())
-		}
-		sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), handler, opts...)
-		if err != nil {
-			return err
-		}
-		g.subscribers[sb] = []broker.Subscriber{sub}
-	}
-
 	g.registered = true
 	g.rsvc = service
 
 	return nil
 }
 
-func (g *grpcServer) Deregister() error {
+func (g *Server) Deregister() error {
 	var err error
 
 	g.RLock()
@@ -792,7 +778,7 @@ func (g *grpcServer) Deregister() error {
 	return nil
 }
 
-func (g *grpcServer) Start() error {
+func (g *Server) Start() error {
 	g.RLock()
 	if g.started {
 		g.RUnlock()
@@ -802,10 +788,6 @@ func (g *grpcServer) Start() error {
 
 	config := g.Options()
 
-	for _, k := range config.Codecs {
-		encoding.RegisterCodec(&wrapMicroCodec{k})
-	}
-
 	// micro: config.Transport.Listen(config.Address)
 	var ts net.Listener
 	var err error
@@ -869,6 +851,10 @@ func (g *grpcServer) Start() error {
 		}
 	}
 
+	if err = g.subscribe(); err != nil {
+		return err
+	}
+
 	// micro: go ts.Accept(s.accept)
 	go func() {
 		if err = g.srv.Serve(ts); err != nil {
@@ -980,7 +966,38 @@ func (g *grpcServer) Start() error {
 	return nil
 }
 
-func (g *grpcServer) Stop() error {
+func (g *Server) subscribe() error {
+	config := g.opts
+
+	for sb := range g.subscribers {
+		handler := g.createSubHandler(sb, config)
+		var opts []broker.SubscribeOption
+		if queue := sb.Options().Queue; len(queue) > 0 {
+			opts = append(opts, broker.SubscribeGroup(queue))
+		}
+
+		subCtx := config.Context
+		if cx := sb.Options().Context; cx != nil {
+			subCtx = cx
+		}
+		opts = append(opts, broker.SubscribeContext(subCtx))
+		opts = append(opts, broker.SubscribeAutoAck(sb.Options().AutoAck))
+		opts = append(opts, broker.SubscribeBodyOnly(sb.Options().BodyOnly))
+
+		if config.Logger.V(logger.InfoLevel) {
+			config.Logger.Infof(config.Context, "Subscribing to topic: %s", sb.Topic())
+		}
+		sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), handler, opts...)
+		if err != nil {
+			return err
+		}
+		g.subscribers[sb] = []broker.Subscriber{sub}
+	}
+
+	return nil
+}
+
+func (g *Server) Stop() error {
 	g.RLock()
 	if !g.started {
 		g.RUnlock()
@@ -1000,14 +1017,18 @@ func (g *grpcServer) Stop() error {
 	return err
 }
 
-func (g *grpcServer) String() string {
+func (g *Server) String() string {
 	return "grpc"
 }
 
-func (g *grpcServer) Name() string {
+func (g *Server) Name() string {
 	return g.opts.Name
 }
 
-func NewServer(opts ...server.Option) server.Server {
-	return newGRPCServer(opts...)
+func (g *Server) GRPCServer() *grpc.Server {
+	return g.srv
+}
+
+func NewServer(opts ...server.Option) *Server {
+	return newServer(opts...)
 }

request.go

@@ -31,7 +31,6 @@ type rpcMessage struct {
 	header      metadata.Metadata
 	topic       string
 	contentType string
-	body        []byte
 }
 
 func (r *rpcRequest) ContentType() string {

subscriber.go

@@ -102,7 +102,7 @@ func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOptio
 	}
 }
 
-func (g *grpcServer) createSubHandler(sb *subscriber, opts server.Options) broker.Handler {
+func (g *Server) createSubHandler(sb *subscriber, opts server.Options) broker.Handler {
 	return func(p broker.Event) (err error) {
 		defer func() {
 			if r := recover(); r != nil {