unistack-org/micro-client-grpc
4
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases 40 Wiki Activity

prepare v4
Some checks failed
lint / lint (pull_request) Failing after 1m41s
Details
test / test (pull_request) Successful in 3m19s
Details

Browse Source
This commit is contained in:
Денис Евстигнеев
2025-02-25 16:42:21 +03:00
parent 9526345cd6
commit 7969cfd8c9
10 changed files with 118 additions and 1198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
grpc.go
View File

@@ -6,18 +6,16 @@ import (
"crypto/tls"
"fmt"
"net"
"os"
"reflect"
"strings"
"sync"
"time"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/selector"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/selector"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
@@ -68,15 +66,15 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
}
func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string
var header map[string][]string
if md, ok := metadata.FromOutgoingContext(ctx); ok {
header = make(map[string]string, len(md))
header = make(map[string][]string, len(md))
for k, v := range md {
header[strings.ToLower(k)] = v
}
} else {
header = make(map[string]string, 2)
header = make(map[string][]string, 2)
}
if opts.RequestMetadata != nil {
for k, v := range opts.RequestMetadata {
@@ -84,16 +82,16 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
}
}
// set timeout in nanoseconds
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
header["timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
header["content-type"] = req.ContentType()
header["Grpc-Timeout"] = append(header["Grpc-Timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
header["timeout"] = append(header["timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
header["content-type"] = append(header["content-type"], req.ContentType())
md := gmetadata.New(header)
md := gmetadata.MD(metadata.Copy(header))
ctx = gmetadata.NewOutgoingContext(ctx, md)
cf, err := g.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
maxRecvMsgSize := g.maxRecvMsgSizeValue()
@@ -134,7 +132,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
return errors.InternalServerError("go.micro.client", "Error sending request: %v", err)
}
defer func() {
// defer execution of release
@@ -180,31 +178,31 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
}
func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string
var header map[string][]string
if md, ok := metadata.FromOutgoingContext(ctx); ok {
header = make(map[string]string, len(md))
header = make(map[string][]string, len(md))
for k, v := range md {
header[k] = v
}
} else {
header = make(map[string]string)
header = make(map[string][]string)
}
// set timeout in nanoseconds
if opts.StreamTimeout > time.Duration(0) {
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
header["timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
header["Grpc-Timeout"] = append(header["Grpc-Timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
header["timeout"] = append(header["timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
}
// set the content type for the request
header["content-type"] = req.ContentType()
header["content-type"] = append(header["content-type"], req.ContentType())
md := gmetadata.New(header)
md := gmetadata.MD(metadata.Copy(header))
ctx = gmetadata.NewOutgoingContext(ctx, md)
cf, err := g.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
var dialCtx context.Context
@@ -245,7 +243,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
return errors.InternalServerError("go.micro.client", "Error sending request: %v", err)
}
desc := &grpc.StreamDesc{
@@ -278,7 +276,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
// release the connection
g.pool.Put(cc, err)
// now return the error
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
return errors.InternalServerError("go.micro.client", "Error creating stream: %v", err)
}
// set request codec
@@ -418,9 +416,6 @@ func (g *grpcClient) Init(opts ...client.Option) error {
if err := g.opts.Meter.Init(); err != nil {
return err
}
if err := g.opts.Transport.Init(); err != nil {
return err
}
g.init = true
@@ -431,10 +426,6 @@ func (g *grpcClient) Options() client.Options {
return g.opts
}
func (g *grpcClient) NewMessage(topic string, msg interface{}, opts ...client.MessageOption) client.Message {
return newGRPCEvent(topic, msg, g.opts.ContentType, opts...)
}
func (g *grpcClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
return newGRPCRequest(service, method, req, g.opts.ContentType, reqOpts...)
}
@@ -476,11 +467,6 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
// make copy of call method
gcall := g.call
// wrap the call in reverse
for i := len(callOpts.CallWrappers); i > 0; i-- {
gcall = callOpts.CallWrappers[i-1](gcall)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = g.opts.Router
@@ -502,7 +488,7 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
// only sleep if greater than 0
@@ -517,7 +503,7 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
// TODO apply any filtering here
routes, err = g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return errors.InternalServerError("go.micro.client", "%+v", err)
}
// balance the list of nodes
@@ -598,11 +584,6 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
// make a copy of stream
gstream := g.stream
// wrap the call in reverse
for i := len(callOpts.CallWrappers); i > 0; i-- {
gstream = callOpts.CallWrappers[i-1](gstream)
}
// use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil {
callOpts.Router = g.opts.Router
@@ -624,7 +605,7 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
return nil, errors.InternalServerError("go.micro.client", "%+v", err)
}
// only sleep if greater than 0
@@ -639,7 +620,7 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
// 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())
return nil, errors.InternalServerError("go.micro.client", "%+v", err)
}
// balance the list of nodes
@@ -712,6 +693,7 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
return nil, grr
}
/*
func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
return g.publish(ctx, ps, opts...)
}
@@ -777,6 +759,8 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
)
}
*/
func (g *grpcClient) String() string {
return "grpc"
}