Merge pull request #138 from unistack-org/dependabot/go_modules/go.unistack.org/micro/v3-3.10.22

Bump go.unistack.org/micro/v3 from 3.10.20 to 3.10.22

.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,13 +10,13 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.16
+        go-version: 1.17
     - name: checkout
       uses: actions/checkout@v3
     - name: cache
-      uses: actions/cache@v2
+      uses: actions/cache@v3
       with:
         path: ~/go/pkg/mod
         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
@@ -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.16
+        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,13 +10,13 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: setup
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
-        go-version: 1.16
+        go-version: 1.17
     - name: checkout
       uses: actions/checkout@v3
     - name: cache
-      uses: actions/cache@v2
+      uses: actions/cache@v3
       with:
         path: ~/go/pkg/mod
         key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
@@ -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.

error.go

@@ -9,36 +9,46 @@ func microError(err error) error {
 	// no error
 
 	if err == nil {
+		// nothing to do
 		return nil
 	}
 
 	if verr, ok := err.(*errors.Error); ok {
+		// micro error
 		return verr
 	}
 
 	// grpc error
 	s, ok := status.FromError(err)
 	if !ok {
+		// can't get status detals from grpc error, return base error
 		return err
 	}
 
-	// return first error from details
+	details := s.Details()
-	if details := s.Details(); len(details) > 0 {
+	switch len(details) {
-		if verr, ok := details[0].(error); ok {
+	case 0:
-			return microError(verr)
+		if verr := errors.Parse(s.Message()); verr.Code > 0 {
+			// return micro error
+			return verr
 		}
+		// return base error as it not micro error
+		return err
+	case 1:
+		if verr, ok := details[0].(*errors.Error); ok {
+			// return nested micro error
+			return verr
+		}
+		// return base error as it not holds micro error
+		return err
 	}
 
-	// try to decode micro *errors.Error
+	// attached messages in details more then 1, try to fallback to micro error
-	if e := errors.Parse(s.Message()); e.Code > 0 {
+	if verr := errors.Parse(s.Message()); verr.Code > 0 {
-		return e // actually a micro error
+		// return micro error
+		return verr
 	}
 
-	// fallback
+	// not micro error return base error
-	return &errors.Error{
+	return err
-		ID:     "go.micro.client",
-		Code:   int32(s.Code()),
-		Detail: s.Message(),
-		Status: s.Code().String(),
-	}
 }

go.mod

@@ -3,10 +3,6 @@ module go.unistack.org/micro-client-grpc/v3
 go 1.16
 
 require (
-	go.unistack.org/micro/v3 v3.8.21
+	go.unistack.org/micro/v3 v3.10.22
-	golang.org/x/net v0.0.0-20220111093109-d55c255bac03 // indirect
+	google.golang.org/grpc v1.52.3
-	golang.org/x/sys v0.0.0-20220111092808-5a964db01320 // indirect
-	golang.org/x/text v0.3.7 // indirect
-	google.golang.org/genproto v0.0.0-20220111164026-67b88f271998 // indirect
-	google.golang.org/grpc v1.44.0
 )

grpc.go

@@ -17,14 +17,16 @@ import (
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/errors"
 	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/selector"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding"
 	gmetadata "google.golang.org/grpc/metadata"
 )
 
 const (
-	defaultContentType = "application/grpc+proto"
+	DefaultContentType = "application/grpc"
 )
 
 type grpcClient struct {
@@ -62,7 +64,7 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
 	}
 
 	// other fallback to insecure
-	return grpc.WithInsecure()
+	return grpc.WithTransportCredentials(insecure.NewCredentials())
 }
 
 func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
@@ -76,9 +78,15 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
 	} else {
 		header = make(map[string]string, 2)
 	}
-
+	if opts.RequestMetadata != nil {
+		for k, v := range opts.RequestMetadata {
+			header[k] = v
+		}
+	}
 	// set timeout in nanoseconds
-	header["timeout"] = fmt.Sprintf("%d", opts.RequestTimeout)
+	header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
+	header["timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
+	header["content-type"] = req.ContentType()
 
 	md := gmetadata.New(header)
 	ctx = gmetadata.NewOutgoingContext(ctx, md)
@@ -110,10 +118,18 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
 		),
 	}
 
-	if opts := g.getGrpcDialOptions(); opts != nil {
+	if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
 		grpcDialOptions = append(grpcDialOptions, opts...)
 	}
 
+	contextDialer := g.opts.ContextDialer
+	if opts.ContextDialer != nil {
+		contextDialer = opts.ContextDialer
+	}
+	if contextDialer != nil {
+		grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
+	}
+
 	cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
@@ -124,15 +140,22 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
 	}()
 
 	ch := make(chan error, 1)
+	var gmd gmetadata.MD
 
-	go func() {
 	grpcCallOptions := []grpc.CallOption{
-			grpc.ForceCodec(&wrapMicroCodec{cf}),
 		grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name()),
 	}
-		if opts := g.getGrpcCallOptions(); opts != nil {
+
+	if opts := g.getGrpcCallOptions(opts.Context); opts != nil {
 		grpcCallOptions = append(grpcCallOptions, opts...)
 	}
+
+	if opts.ResponseMetadata != nil {
+		gmd = gmetadata.MD{}
+		grpcCallOptions = append(grpcCallOptions, grpc.Header(&gmd))
+	}
+
+	go func() {
 		err := cc.Invoke(ctx, methodToGRPC(req.Service(), req.Endpoint()), req.Body(), rsp, grpcCallOptions...)
 		ch <- microError(err)
 	}()
@@ -144,6 +167,13 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
 		grr = errors.Timeout("go.micro.client", "%v", ctx.Err())
 	}
 
+	if opts.ResponseMetadata != nil {
+		*opts.ResponseMetadata = metadata.New(gmd.Len())
+		for k, v := range gmd {
+			opts.ResponseMetadata.Set(k, strings.Join(v, ","))
+		}
+	}
+
 	return grr
 }
 
@@ -161,10 +191,11 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 
 	// set timeout in nanoseconds
 	if opts.StreamTimeout > time.Duration(0) {
-		header["timeout"] = fmt.Sprintf("%d", opts.StreamTimeout)
+		header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
+		header["timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
 	}
 	// set the content type for the request
-	header["x-content-type"] = req.ContentType()
+	header["content-type"] = req.ContentType()
 
 	md := gmetadata.New(header)
 	ctx = gmetadata.NewOutgoingContext(ctx, md)
@@ -196,10 +227,18 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 		),
 	}
 
-	if opts := g.getGrpcDialOptions(); opts != nil {
+	if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
 		grpcDialOptions = append(grpcDialOptions, opts...)
 	}
 
+	contextDialer := g.opts.ContextDialer
+	if opts.ContextDialer != nil {
+		contextDialer = opts.ContextDialer
+	}
+	if contextDialer != nil {
+		grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
+	}
+
 	cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
@@ -212,12 +251,17 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 	}
 
 	grpcCallOptions := []grpc.CallOption{
-		grpc.ForceCodec(wc),
+		// grpc.ForceCodec(wc),
 		grpc.CallContentSubtype(wc.Name()),
 	}
-	if opts := g.getGrpcCallOptions(); opts != nil {
+	if opts := g.getGrpcCallOptions(opts.Context); opts != nil {
 		grpcCallOptions = append(grpcCallOptions, opts...)
 	}
+	var gmd gmetadata.MD
+	if opts.ResponseMetadata != nil {
+		gmd = gmetadata.MD{}
+		grpcCallOptions = append(grpcCallOptions, grpc.Header(&gmd))
+	}
 
 	// create a new cancelling context
 	newCtx, cancel := context.WithCancel(ctx)
@@ -263,6 +307,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 	// set the stream as the response
 	val := reflect.ValueOf(rsp).Elem()
 	val.Set(reflect.ValueOf(stream).Elem())
+
 	return nil
 }
 
@@ -387,6 +432,7 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 	}
 	// make a copy of call opts
 	callOpts := g.opts.CallOptions
+
 	for _, opt := range opts {
 		opt(&callOpts)
 	}
@@ -435,20 +481,8 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 		callOpts.Address = []string{g.opts.Proxy}
 	}
 
-	// lookup the route to send the reques to
+	var next selector.Next
-	// TODO apply any filtering here
-	routes, err := g.opts.Lookup(ctx, req, callOpts)
-	if err != nil {
-		return errors.InternalServerError("go.micro.client", err.Error())
-	}
 
-	// balance the list of nodes
-	next, err := callOpts.Selector.Select(routes)
-	if err != nil {
-		return err
-	}
-
-	// return errors.New("go.micro.client", "request timeout", 408)
 	call := func(i int) error {
 		// call backoff first. Someone may want an initial start delay
 		t, err := callOpts.Backoff(ctx, req, i)
@@ -461,6 +495,23 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 			time.Sleep(t)
 		}
 
+		if next == nil {
+			var routes []string
+
+			// lookup the route to send the reques to
+			// TODO apply any filtering here
+			routes, err = g.opts.Lookup(ctx, req, callOpts)
+			if err != nil {
+				return errors.InternalServerError("go.micro.client", err.Error())
+			}
+
+			// balance the list of nodes
+			next, err = callOpts.Selector.Select(routes)
+			if err != nil {
+				return err
+			}
+		}
+
 		// get the next node
 		node := next()
 
@@ -552,18 +603,7 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 		callOpts.Address = []string{g.opts.Proxy}
 	}
 
-	// lookup the route to send the reques to
+	var next selector.Next
-	// TODO: move to internal lookup func
-	routes, err := g.opts.Lookup(ctx, req, callOpts)
-	if err != nil {
-		return nil, errors.InternalServerError("go.micro.client", err.Error())
-	}
-
-	// balance the list of nodes
-	next, err := callOpts.Selector.Select(routes)
-	if err != nil {
-		return nil, err
-	}
 
 	call := func(i int) (client.Stream, error) {
 		// call backoff first. Someone may want an initial start delay
@@ -577,6 +617,23 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 			time.Sleep(t)
 		}
 
+		if next == nil {
+			var routes []string
+
+			// lookup the route to send the reques to
+			// TODO apply any filtering here
+			routes, err = g.opts.Lookup(ctx, req, callOpts)
+			if err != nil {
+				return nil, errors.InternalServerError("go.micro.client", err.Error())
+			}
+
+			// balance the list of nodes
+			next, err = callOpts.Selector.Select(routes)
+			if err != nil {
+				return nil, err
+			}
+		}
+
 		// get the next node
 		node := next()
 
@@ -713,41 +770,45 @@ func (g *grpcClient) Name() string {
 	return g.opts.Name
 }
 
-func (g *grpcClient) getGrpcDialOptions() []grpc.DialOption {
+func (g *grpcClient) getGrpcDialOptions(ctx context.Context) []grpc.DialOption {
-	if g.opts.CallOptions.Context == nil {
+	var opts []grpc.DialOption
-		return nil
+
+	if g.opts.CallOptions.Context != nil {
+		if v := g.opts.CallOptions.Context.Value(grpcDialOptions{}); v != nil {
+			if vopts, ok := v.([]grpc.DialOption); ok {
+				opts = append(opts, vopts...)
+			}
+		}
 	}
 
-	v := g.opts.CallOptions.Context.Value(grpcDialOptions{})
+	if ctx != nil {
-
+		if v := ctx.Value(grpcDialOptions{}); v != nil {
-	if v == nil {
+			if vopts, ok := v.([]grpc.DialOption); ok {
-		return nil
+				opts = append(opts, vopts...)
+			}
 		}
-
-	opts, ok := v.([]grpc.DialOption)
-
-	if !ok {
-		return nil
 	}
 
 	return opts
 }
 
-func (g *grpcClient) getGrpcCallOptions() []grpc.CallOption {
+func (g *grpcClient) getGrpcCallOptions(ctx context.Context) []grpc.CallOption {
-	if g.opts.CallOptions.Context == nil {
+	var opts []grpc.CallOption
-		return nil
+
+	if g.opts.CallOptions.Context != nil {
+		if v := g.opts.CallOptions.Context.Value(grpcCallOptions{}); v != nil {
+			if vopts, ok := v.([]grpc.CallOption); ok {
+				opts = append(opts, vopts...)
+			}
+		}
 	}
 
-	v := g.opts.CallOptions.Context.Value(grpcCallOptions{})
+	if ctx != nil {
-
+		if v := ctx.Value(grpcCallOptions{}); v != nil {
-	if v == nil {
+			if vopts, ok := v.([]grpc.CallOption); ok {
-		return nil
+				opts = append(opts, vopts...)
+			}
 		}
-
-	opts, ok := v.([]grpc.CallOption)
-
-	if !ok {
-		return nil
 	}
 
 	return opts
@@ -756,7 +817,9 @@ func (g *grpcClient) getGrpcCallOptions() []grpc.CallOption {
 func NewClient(opts ...client.Option) client.Client {
 	options := client.NewOptions(opts...)
 	// default content type for grpc
-	options.ContentType = defaultContentType
+	if options.ContentType == "" {
+		options.ContentType = DefaultContentType
+	}
 
 	rc := &grpcClient{
 		opts: options,

grpc_pool.go

@@ -2,6 +2,7 @@ package grpc
 
 import (
 	"context"
+	"strings"
 	"sync"
 	"time"
 
@@ -59,6 +60,9 @@ func newPool(size int, ttl time.Duration, idle int, ms int) *pool {
 }
 
 func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption) (*poolConn, error) {
+	if strings.HasPrefix(addr, "http") {
+		addr = addr[strings.Index(addr, ":")+3:]
+	}
 	now := time.Now().Unix()
 	p.Lock()
 	sp, ok := p.conns[addr]
@@ -126,7 +130,7 @@ func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption
 	}
 	p.Unlock()
 
-	//  create new conn
+	//  create new conn)
 	cc, err := grpc.DialContext(ctx, addr, opts...)
 	if err != nil {
 		return nil, err

options.go

@@ -70,9 +70,7 @@ func Codec(contentType string, c encoding.Codec) client.Option {
 
 type maxRecvMsgSizeKey struct{}
 
-//
 // MaxRecvMsgSize set the maximum size of message that client can receive.
-//
 func MaxRecvMsgSize(s int) client.Option {
 	return func(o *client.Options) {
 		if o.Context == nil {
@@ -84,9 +82,7 @@ func MaxRecvMsgSize(s int) client.Option {
 
 type maxSendMsgSizeKey struct{}
 
-//
 // MaxSendMsgSize set the maximum size of message that client can send.
-//
 func MaxSendMsgSize(s int) client.Option {
 	return func(o *client.Options) {
 		if o.Context == nil {
@@ -98,9 +94,7 @@ func MaxSendMsgSize(s int) client.Option {
 
 type grpcDialOptions struct{}
 
-//
 // DialOptions to be used to configure gRPC dial options
-//
 func DialOptions(opts ...grpc.DialOption) client.CallOption {
 	return func(o *client.CallOptions) {
 		if o.Context == nil {
@@ -112,9 +106,7 @@ func DialOptions(opts ...grpc.DialOption) client.CallOption {
 
 type grpcCallOptions struct{}
 
-//
 // CallOptions to be used to configure gRPC call options
-//
 func CallOptions(opts ...grpc.CallOption) client.CallOption {
 	return func(o *client.CallOptions) {
 		if o.Context == nil {

request.go

@@ -38,15 +38,12 @@ func methodToGRPC(service, method string) string {
 	return fmt.Sprintf("/%s.%s/%s", service, mParts[0], mParts[1])
 }
 
-func newGRPCRequest(service, method string, request interface{}, contentType string, reqOpts ...client.RequestOption) client.Request {
+func newGRPCRequest(service, method string, request interface{}, contentType string, opts ...client.RequestOption) client.Request {
-	var opts client.RequestOptions
+	options := client.NewRequestOptions(opts...)
-	for _, o := range reqOpts {
-		o(&opts)
-	}
 
 	// set the content-type specified
-	if len(opts.ContentType) > 0 {
+	if len(options.ContentType) > 0 {
-		contentType = opts.ContentType
+		contentType = options.ContentType
 	}
 
 	return &grpcRequest{
@@ -54,7 +51,7 @@ func newGRPCRequest(service, method string, request interface{}, contentType str
 		method:      method,
 		request:     request,
 		contentType: contentType,
-		opts:        opts,
+		opts:        options,
 	}
 }
 

response.go

@@ -23,7 +23,7 @@ func (r *response) Codec() codec.Codec {
 func (r *response) Header() metadata.Metadata {
 	meta, err := r.stream.Header()
 	if err != nil {
-		return metadata.New(0)
+		return nil
 	}
 	md := metadata.New(len(meta))
 	for k, v := range meta {