Compare commits
67 Commits
v3.8.6
...
40b5402aa5
| Author | SHA1 | Date | |
|---|---|---|---|
| 40b5402aa5 | |||
|
|
e8b7e30e4d | ||
| 55e7e3d61d | |||
|
|
d75c36938a | ||
| a7cec360e9 | |||
|
|
3978256931 | ||
| 6dfff28203 | |||
|
|
373acda151 | ||
| 3f9ed4e83b | |||
|
|
0486780f36 | ||
| 36788b5fbf | |||
|
|
c79b83a3c1 | ||
| a6585fd6d4 | |||
|
|
7fce5ccad9 | ||
| b5a5e98f9e | |||
|
|
e419c6a67c | ||
| 17c46e63a2 | |||
|
|
cdf1a2c3e3 | ||
| aba3df63df | |||
|
|
337df46029 | ||
| bfe4839a00 | |||
|
|
5474c37f8f | ||
| ba88a5568a | |||
|
|
57f903a8c2 | ||
| 9f918bd3f2 | |||
|
|
0d88050daf | ||
|
|
d1ed5bed51 | ||
|
|
42697318b1 | ||
| 28aea45725 | |||
|
|
ed72c05645 | ||
|
|
d239dcde63 | ||
| 6eb07dc351 | |||
|
|
a0d704f845 | ||
| 4b8a761f30 | |||
|
|
a2c711a1b5 | ||
| be564f50aa | |||
|
|
b7b1eff81c | ||
| 671a6b6f7c | |||
| 27c1c4d86b | |||
| 76d37a99eb | |||
|
|
4c2827172f | ||
| e3461dd23f | |||
|
|
d8b5c011e5 | ||
|
|
15abd38afd | ||
| 5865a0f388 | |||
|
|
67da8d1165 | ||
|
|
c7d24caa03 | ||
| 8c222c4715 | |||
| 6e91cd5cf5 | |||
| a9e673b2ef | |||
| 251f06cc31 | |||
| e97e4580a1 | |||
| bd7dbe94ca | |||
|
|
0f32fad4c0 | ||
|
|
858111106e | ||
|
|
dc35dc6d3e | ||
|
|
24b1abfb9a | ||
|
|
205fd53047 | ||
|
|
b3b7d1af13 | ||
| 80e2184bba | |||
|
|
a920d15d95 | ||
| 5ecbfac164 | |||
|
|
61d7a322de | ||
|
|
b43c207f6a | ||
|
|
d243b884c7 | ||
| 107b470b9a | |||
|
|
878bac53ac |
2
.github/workflows/autoapprove.yml
vendored
2
.github/workflows/autoapprove.yml
vendored
@@ -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:
|
||||
|
||||
2
.github/workflows/build.yml
vendored
2
.github/workflows/build.yml
vendored
@@ -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.
|
||||
|
||||
2
.github/workflows/dependabot-automerge.yml
vendored
2
.github/workflows/dependabot-automerge.yml
vendored
@@ -15,7 +15,7 @@ jobs:
|
||||
steps:
|
||||
- name: metadata
|
||||
id: metadata
|
||||
uses: dependabot/fetch-metadata@v1.3.1
|
||||
uses: dependabot/fetch-metadata@v1.3.6
|
||||
with:
|
||||
github-token: "${{ secrets.TOKEN }}"
|
||||
- name: merge
|
||||
|
||||
2
.github/workflows/pr.yml
vendored
2
.github/workflows/pr.yml
vendored
@@ -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.
|
||||
|
||||
12
go.mod
12
go.mod
@@ -3,14 +3,6 @@ module go.unistack.org/micro-client-grpc/v3
|
||||
go 1.16
|
||||
|
||||
require (
|
||||
github.com/google/gnostic v0.6.9 // indirect
|
||||
github.com/google/go-cmp v0.5.7 // indirect
|
||||
github.com/kr/pretty v0.2.1 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
go.unistack.org/micro/v3 v3.9.7
|
||||
golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 // indirect
|
||||
golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32 // indirect
|
||||
google.golang.org/genproto v0.0.0-20220429170224-98d788798c3e // indirect
|
||||
google.golang.org/grpc v1.46.0
|
||||
google.golang.org/protobuf v1.28.0 // indirect
|
||||
go.unistack.org/micro/v3 v3.10.22
|
||||
google.golang.org/grpc v1.52.3
|
||||
)
|
||||
|
||||
130
grpc.go
130
grpc.go
@@ -26,7 +26,7 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
DefaultContentType = "application/grpc+proto"
|
||||
DefaultContentType = "application/grpc"
|
||||
)
|
||||
|
||||
type grpcClient struct {
|
||||
@@ -78,10 +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["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)
|
||||
@@ -113,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))
|
||||
@@ -127,15 +140,22 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
|
||||
}()
|
||||
|
||||
ch := make(chan error, 1)
|
||||
var gmd gmetadata.MD
|
||||
|
||||
grpcCallOptions := []grpc.CallOption{
|
||||
grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name()),
|
||||
}
|
||||
|
||||
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() {
|
||||
grpcCallOptions := []grpc.CallOption{
|
||||
grpc.ForceCodec(&wrapMicroCodec{cf}),
|
||||
grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name()),
|
||||
}
|
||||
if opts := g.getGrpcCallOptions(); opts != nil {
|
||||
grpcCallOptions = append(grpcCallOptions, opts...)
|
||||
}
|
||||
err := cc.Invoke(ctx, methodToGRPC(req.Service(), req.Endpoint()), req.Body(), rsp, grpcCallOptions...)
|
||||
ch <- microError(err)
|
||||
}()
|
||||
@@ -147,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
|
||||
}
|
||||
|
||||
@@ -168,7 +195,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
|
||||
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)
|
||||
@@ -200,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))
|
||||
@@ -216,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)
|
||||
@@ -267,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
|
||||
}
|
||||
|
||||
@@ -391,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)
|
||||
}
|
||||
@@ -728,41 +770,45 @@ func (g *grpcClient) Name() string {
|
||||
return g.opts.Name
|
||||
}
|
||||
|
||||
func (g *grpcClient) getGrpcDialOptions() []grpc.DialOption {
|
||||
if g.opts.CallOptions.Context == nil {
|
||||
return nil
|
||||
func (g *grpcClient) getGrpcDialOptions(ctx context.Context) []grpc.DialOption {
|
||||
var opts []grpc.DialOption
|
||||
|
||||
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 v == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
opts, ok := v.([]grpc.DialOption)
|
||||
|
||||
if !ok {
|
||||
return nil
|
||||
if ctx != nil {
|
||||
if v := ctx.Value(grpcDialOptions{}); v != nil {
|
||||
if vopts, ok := v.([]grpc.DialOption); ok {
|
||||
opts = append(opts, vopts...)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return opts
|
||||
}
|
||||
|
||||
func (g *grpcClient) getGrpcCallOptions() []grpc.CallOption {
|
||||
if g.opts.CallOptions.Context == nil {
|
||||
return nil
|
||||
func (g *grpcClient) getGrpcCallOptions(ctx context.Context) []grpc.CallOption {
|
||||
var opts []grpc.CallOption
|
||||
|
||||
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 v == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
opts, ok := v.([]grpc.CallOption)
|
||||
|
||||
if !ok {
|
||||
return nil
|
||||
if ctx != nil {
|
||||
if v := ctx.Value(grpcCallOptions{}); v != nil {
|
||||
if vopts, ok := v.([]grpc.CallOption); ok {
|
||||
opts = append(opts, vopts...)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return opts
|
||||
@@ -771,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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
13
request.go
13
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 {
|
||||
var opts client.RequestOptions
|
||||
for _, o := range reqOpts {
|
||||
o(&opts)
|
||||
}
|
||||
func newGRPCRequest(service, method string, request interface{}, contentType string, opts ...client.RequestOption) client.Request {
|
||||
options := client.NewRequestOptions(opts...)
|
||||
|
||||
// set the content-type specified
|
||||
if len(opts.ContentType) > 0 {
|
||||
contentType = opts.ContentType
|
||||
if len(options.ContentType) > 0 {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user