2019-06-03 20:44:43 +03:00
|
|
|
// Package grpc provides a gRPC client
|
2021-10-27 00:57:12 +03:00
|
|
|
package grpc // import "go.unistack.org/micro-client-grpc/v3"
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"crypto/tls"
|
|
|
|
"fmt"
|
2020-02-19 11:44:35 +03:00
|
|
|
"net"
|
2021-07-31 15:10:59 +03:00
|
|
|
"os"
|
2020-06-02 19:56:26 +03:00
|
|
|
"reflect"
|
2020-02-19 11:44:35 +03:00
|
|
|
"strings"
|
2020-09-18 18:27:29 +03:00
|
|
|
"sync"
|
2019-06-03 20:44:43 +03:00
|
|
|
"time"
|
|
|
|
|
2021-10-27 00:57:12 +03:00
|
|
|
"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"
|
2022-05-01 23:23:15 +03:00
|
|
|
"go.unistack.org/micro/v3/selector"
|
2019-06-03 20:44:43 +03:00
|
|
|
"google.golang.org/grpc"
|
|
|
|
"google.golang.org/grpc/credentials"
|
2022-03-21 14:09:08 +03:00
|
|
|
"google.golang.org/grpc/credentials/insecure"
|
2019-06-03 20:44:43 +03:00
|
|
|
"google.golang.org/grpc/encoding"
|
|
|
|
gmetadata "google.golang.org/grpc/metadata"
|
|
|
|
)
|
|
|
|
|
2021-05-24 23:11:05 +03:00
|
|
|
const (
|
2023-01-07 23:27:04 +03:00
|
|
|
DefaultContentType = "application/grpc"
|
2021-05-24 23:11:05 +03:00
|
|
|
)
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
type grpcClient struct {
|
2021-04-26 19:17:29 +03:00
|
|
|
pool *pool
|
2021-07-31 15:10:59 +03:00
|
|
|
opts client.Options
|
2020-09-18 18:27:29 +03:00
|
|
|
sync.RWMutex
|
2021-07-31 15:10:59 +03:00
|
|
|
init bool
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// secure returns the dial option for whether its a secure or insecure connection
|
2020-02-19 11:44:35 +03:00
|
|
|
func (g *grpcClient) secure(addr string) grpc.DialOption {
|
|
|
|
// first we check if theres'a tls config
|
2021-04-20 12:21:00 +03:00
|
|
|
if g.opts.TLSConfig != nil {
|
|
|
|
creds := credentials.NewTLS(g.opts.TLSConfig)
|
|
|
|
// return tls config if it exists
|
|
|
|
return grpc.WithTransportCredentials(creds)
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
2020-02-19 11:44:35 +03:00
|
|
|
|
|
|
|
// default config
|
2021-04-26 19:17:29 +03:00
|
|
|
tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12}
|
2020-02-19 11:44:35 +03:00
|
|
|
defaultCreds := grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig))
|
|
|
|
|
|
|
|
// check if the address is prepended with https
|
|
|
|
if strings.HasPrefix(addr, "https://") {
|
|
|
|
return defaultCreds
|
|
|
|
}
|
|
|
|
|
|
|
|
// if no port is specified or port is 443 default to tls
|
|
|
|
_, port, err := net.SplitHostPort(addr)
|
|
|
|
// assuming with no port its going to be secured
|
|
|
|
if port == "443" {
|
|
|
|
return defaultCreds
|
|
|
|
} else if err != nil && strings.Contains(err.Error(), "missing port in address") {
|
|
|
|
return defaultCreds
|
|
|
|
}
|
|
|
|
|
|
|
|
// other fallback to insecure
|
2022-03-21 14:09:08 +03:00
|
|
|
return grpc.WithTransportCredentials(insecure.NewCredentials())
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
2020-08-18 00:44:45 +03:00
|
|
|
func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
|
2020-02-24 17:15:20 +03:00
|
|
|
var header map[string]string
|
|
|
|
|
2021-02-12 20:34:03 +03:00
|
|
|
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
2020-02-24 17:15:20 +03:00
|
|
|
header = make(map[string]string, len(md))
|
2019-06-03 20:44:43 +03:00
|
|
|
for k, v := range md {
|
2020-03-10 19:47:01 +03:00
|
|
|
header[strings.ToLower(k)] = v
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
2020-02-24 17:15:20 +03:00
|
|
|
} else {
|
2020-10-28 10:07:30 +03:00
|
|
|
header = make(map[string]string, 2)
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
2022-11-16 23:10:27 +03:00
|
|
|
if opts.RequestMetadata != nil {
|
|
|
|
for k, v := range opts.RequestMetadata {
|
|
|
|
header[k] = v
|
|
|
|
}
|
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
// set timeout in nanoseconds
|
2022-03-21 14:09:08 +03:00
|
|
|
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
|
|
|
|
header["timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
|
2022-11-16 23:10:27 +03:00
|
|
|
header["content-type"] = req.ContentType()
|
2020-03-10 19:47:01 +03:00
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
md := gmetadata.New(header)
|
|
|
|
ctx = gmetadata.NewOutgoingContext(ctx, md)
|
|
|
|
|
2021-04-16 17:07:04 +03:00
|
|
|
cf, err := g.newCodec(req.ContentType())
|
2019-06-03 20:44:43 +03:00
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
|
|
|
maxSendMsgSize := g.maxSendMsgSizeValue()
|
|
|
|
|
|
|
|
var grr error
|
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
var dialCtx context.Context
|
|
|
|
var cancel context.CancelFunc
|
|
|
|
if opts.DialTimeout >= 0 {
|
|
|
|
dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
|
|
|
|
} else {
|
|
|
|
dialCtx, cancel = context.WithCancel(ctx)
|
|
|
|
}
|
|
|
|
defer cancel()
|
|
|
|
|
2019-10-09 09:45:51 +03:00
|
|
|
grpcDialOptions := []grpc.DialOption{
|
2020-08-18 00:44:45 +03:00
|
|
|
g.secure(addr),
|
2019-06-03 20:44:43 +03:00
|
|
|
grpc.WithDefaultCallOptions(
|
|
|
|
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
|
|
|
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
2019-10-09 09:45:51 +03:00
|
|
|
),
|
|
|
|
}
|
|
|
|
|
2023-06-23 12:23:44 +03:00
|
|
|
if opts := g.getGrpcDialOptions(g.opts.Context); opts != nil {
|
|
|
|
grpcDialOptions = append(grpcDialOptions, opts...)
|
|
|
|
}
|
2022-11-16 23:10:27 +03:00
|
|
|
if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
|
2019-10-09 09:45:51 +03:00
|
|
|
grpcDialOptions = append(grpcDialOptions, opts...)
|
|
|
|
}
|
|
|
|
|
2022-11-16 23:58:24 +03:00
|
|
|
contextDialer := g.opts.ContextDialer
|
|
|
|
if opts.ContextDialer != nil {
|
|
|
|
contextDialer = opts.ContextDialer
|
|
|
|
}
|
|
|
|
if contextDialer != nil {
|
|
|
|
grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
|
|
|
|
}
|
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
|
2019-06-03 20:44:43 +03:00
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
|
|
|
|
}
|
|
|
|
defer func() {
|
|
|
|
// defer execution of release
|
2021-04-26 19:17:29 +03:00
|
|
|
g.pool.release(cc, grr)
|
2019-06-03 20:44:43 +03:00
|
|
|
}()
|
|
|
|
|
|
|
|
ch := make(chan error, 1)
|
2022-11-16 23:10:27 +03:00
|
|
|
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))
|
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
go func() {
|
2019-10-09 09:45:51 +03:00
|
|
|
err := cc.Invoke(ctx, methodToGRPC(req.Service(), req.Endpoint()), req.Body(), rsp, grpcCallOptions...)
|
2019-06-03 20:44:43 +03:00
|
|
|
ch <- microError(err)
|
|
|
|
}()
|
|
|
|
|
|
|
|
select {
|
|
|
|
case err := <-ch:
|
|
|
|
grr = err
|
|
|
|
case <-ctx.Done():
|
2020-02-06 00:13:14 +03:00
|
|
|
grr = errors.Timeout("go.micro.client", "%v", ctx.Err())
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
if opts.ResponseMetadata != nil {
|
|
|
|
*opts.ResponseMetadata = metadata.New(gmd.Len())
|
|
|
|
for k, v := range gmd {
|
|
|
|
opts.ResponseMetadata.Set(k, strings.Join(v, ","))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
return grr
|
|
|
|
}
|
|
|
|
|
2020-08-18 00:44:45 +03:00
|
|
|
func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
|
2020-02-24 17:15:20 +03:00
|
|
|
var header map[string]string
|
|
|
|
|
2021-02-12 20:34:03 +03:00
|
|
|
if md, ok := metadata.FromOutgoingContext(ctx); ok {
|
2020-02-24 17:15:20 +03:00
|
|
|
header = make(map[string]string, len(md))
|
2019-06-03 20:44:43 +03:00
|
|
|
for k, v := range md {
|
|
|
|
header[k] = v
|
|
|
|
}
|
2020-02-24 17:15:20 +03:00
|
|
|
} else {
|
|
|
|
header = make(map[string]string)
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// set timeout in nanoseconds
|
2020-04-01 01:22:11 +03:00
|
|
|
if opts.StreamTimeout > time.Duration(0) {
|
2022-03-21 14:09:08 +03:00
|
|
|
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
|
|
|
|
header["timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
|
2020-04-01 01:22:11 +03:00
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
// set the content type for the request
|
2022-11-16 23:10:27 +03:00
|
|
|
header["content-type"] = req.ContentType()
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
md := gmetadata.New(header)
|
|
|
|
ctx = gmetadata.NewOutgoingContext(ctx, md)
|
|
|
|
|
2021-04-16 17:07:04 +03:00
|
|
|
cf, err := g.newCodec(req.ContentType())
|
2019-06-03 20:44:43 +03:00
|
|
|
if err != nil {
|
2020-06-02 19:56:26 +03:00
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
var dialCtx context.Context
|
|
|
|
var cancel context.CancelFunc
|
|
|
|
if opts.DialTimeout >= 0 {
|
|
|
|
dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
|
|
|
|
} else {
|
|
|
|
dialCtx, cancel = context.WithCancel(ctx)
|
|
|
|
}
|
|
|
|
defer cancel()
|
|
|
|
|
2020-11-26 01:18:35 +03:00
|
|
|
wc := &wrapMicroCodec{cf}
|
2019-06-17 22:05:58 +03:00
|
|
|
|
2021-06-05 16:24:29 +03:00
|
|
|
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
|
|
|
maxSendMsgSize := g.maxSendMsgSizeValue()
|
|
|
|
|
2019-10-09 09:45:51 +03:00
|
|
|
grpcDialOptions := []grpc.DialOption{
|
2020-08-18 00:44:45 +03:00
|
|
|
g.secure(addr),
|
2021-06-05 16:24:29 +03:00
|
|
|
grpc.WithDefaultCallOptions(
|
|
|
|
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
|
|
|
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
|
|
|
),
|
2019-10-09 09:45:51 +03:00
|
|
|
}
|
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
|
2019-10-09 09:45:51 +03:00
|
|
|
grpcDialOptions = append(grpcDialOptions, opts...)
|
|
|
|
}
|
|
|
|
|
2022-11-17 00:00:50 +03:00
|
|
|
contextDialer := g.opts.ContextDialer
|
|
|
|
if opts.ContextDialer != nil {
|
|
|
|
contextDialer = opts.ContextDialer
|
|
|
|
}
|
|
|
|
if contextDialer != nil {
|
|
|
|
grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
|
|
|
|
}
|
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
|
2019-06-03 20:44:43 +03:00
|
|
|
if err != nil {
|
2020-06-02 19:56:26 +03:00
|
|
|
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
desc := &grpc.StreamDesc{
|
|
|
|
StreamName: req.Service() + req.Endpoint(),
|
|
|
|
ClientStreams: true,
|
|
|
|
ServerStreams: true,
|
|
|
|
}
|
|
|
|
|
2020-05-01 16:59:50 +03:00
|
|
|
grpcCallOptions := []grpc.CallOption{
|
2022-11-16 23:10:27 +03:00
|
|
|
// grpc.ForceCodec(wc),
|
2020-11-26 01:18:35 +03:00
|
|
|
grpc.CallContentSubtype(wc.Name()),
|
2020-05-01 16:59:50 +03:00
|
|
|
}
|
2022-11-16 23:10:27 +03:00
|
|
|
if opts := g.getGrpcCallOptions(opts.Context); opts != nil {
|
2019-10-09 09:45:51 +03:00
|
|
|
grpcCallOptions = append(grpcCallOptions, opts...)
|
|
|
|
}
|
2022-11-16 23:10:27 +03:00
|
|
|
var gmd gmetadata.MD
|
|
|
|
if opts.ResponseMetadata != nil {
|
|
|
|
gmd = gmetadata.MD{}
|
|
|
|
grpcCallOptions = append(grpcCallOptions, grpc.Header(&gmd))
|
|
|
|
}
|
2020-01-09 20:00:14 +03:00
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
// create a new cancelling context
|
|
|
|
newCtx, cancel := context.WithCancel(ctx)
|
2020-01-20 01:53:56 +03:00
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
st, err := cc.NewStream(newCtx, desc, methodToGRPC(req.Service(), req.Endpoint()), grpcCallOptions...)
|
2019-06-03 20:44:43 +03:00
|
|
|
if err != nil {
|
2020-02-15 14:35:08 +03:00
|
|
|
// we need to cleanup as we dialled and created a context
|
|
|
|
// cancel the context
|
|
|
|
cancel()
|
2020-09-17 16:08:21 +03:00
|
|
|
// release the connection
|
2021-04-26 19:17:29 +03:00
|
|
|
g.pool.release(cc, err)
|
2020-02-15 14:35:08 +03:00
|
|
|
// now return the error
|
2020-06-02 19:56:26 +03:00
|
|
|
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
2019-06-17 22:05:58 +03:00
|
|
|
// set request codec
|
|
|
|
if r, ok := req.(*grpcRequest); ok {
|
2020-11-26 01:18:35 +03:00
|
|
|
r.codec = cf
|
2019-06-17 22:05:58 +03:00
|
|
|
}
|
|
|
|
|
2020-06-02 19:56:26 +03:00
|
|
|
// setup the stream response
|
|
|
|
stream := &grpcStream{
|
2020-08-09 17:43:41 +03:00
|
|
|
ClientStream: st,
|
|
|
|
context: ctx,
|
|
|
|
request: req,
|
2020-06-02 19:56:26 +03:00
|
|
|
response: &response{
|
|
|
|
conn: cc,
|
|
|
|
stream: st,
|
|
|
|
codec: cf,
|
|
|
|
},
|
2020-09-17 16:08:21 +03:00
|
|
|
conn: cc,
|
|
|
|
close: func(err error) {
|
2021-04-26 19:17:29 +03:00
|
|
|
// cancel the context if an error occurred
|
2020-09-17 16:08:21 +03:00
|
|
|
if err != nil {
|
|
|
|
cancel()
|
|
|
|
}
|
|
|
|
|
|
|
|
// defer execution of release
|
2021-04-26 19:17:29 +03:00
|
|
|
g.pool.release(cc, err)
|
2020-09-17 16:08:21 +03:00
|
|
|
},
|
2020-06-02 19:56:26 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// set the stream as the response
|
|
|
|
val := reflect.ValueOf(rsp).Elem()
|
|
|
|
val.Set(reflect.ValueOf(stream).Elem())
|
2022-11-16 23:10:27 +03:00
|
|
|
|
2020-06-02 19:56:26 +03:00
|
|
|
return nil
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
2019-12-27 15:25:58 +03:00
|
|
|
func (g *grpcClient) poolMaxStreams() int {
|
|
|
|
if g.opts.Context == nil {
|
|
|
|
return DefaultPoolMaxStreams
|
|
|
|
}
|
|
|
|
v := g.opts.Context.Value(poolMaxStreams{})
|
|
|
|
if v == nil {
|
|
|
|
return DefaultPoolMaxStreams
|
|
|
|
}
|
|
|
|
return v.(int)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) poolMaxIdle() int {
|
|
|
|
if g.opts.Context == nil {
|
|
|
|
return DefaultPoolMaxIdle
|
|
|
|
}
|
|
|
|
v := g.opts.Context.Value(poolMaxIdle{})
|
|
|
|
if v == nil {
|
|
|
|
return DefaultPoolMaxIdle
|
|
|
|
}
|
|
|
|
return v.(int)
|
|
|
|
}
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
func (g *grpcClient) maxRecvMsgSizeValue() int {
|
|
|
|
if g.opts.Context == nil {
|
|
|
|
return DefaultMaxRecvMsgSize
|
|
|
|
}
|
|
|
|
v := g.opts.Context.Value(maxRecvMsgSizeKey{})
|
|
|
|
if v == nil {
|
|
|
|
return DefaultMaxRecvMsgSize
|
|
|
|
}
|
|
|
|
return v.(int)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) maxSendMsgSizeValue() int {
|
|
|
|
if g.opts.Context == nil {
|
|
|
|
return DefaultMaxSendMsgSize
|
|
|
|
}
|
|
|
|
v := g.opts.Context.Value(maxSendMsgSizeKey{})
|
|
|
|
if v == nil {
|
|
|
|
return DefaultMaxSendMsgSize
|
|
|
|
}
|
|
|
|
return v.(int)
|
|
|
|
}
|
|
|
|
|
2021-04-16 17:07:04 +03:00
|
|
|
func (g *grpcClient) newCodec(ct string) (codec.Codec, error) {
|
2020-09-18 18:27:29 +03:00
|
|
|
g.RLock()
|
|
|
|
defer g.RUnlock()
|
|
|
|
|
2021-04-16 17:07:04 +03:00
|
|
|
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
|
|
|
|
ct = ct[:idx]
|
|
|
|
}
|
|
|
|
|
2020-11-26 01:18:35 +03:00
|
|
|
if c, ok := g.opts.Codecs[ct]; ok {
|
|
|
|
return c, nil
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
2020-11-26 01:18:35 +03:00
|
|
|
return nil, codec.ErrUnknownContentType
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) Init(opts ...client.Option) error {
|
2021-03-24 15:32:19 +03:00
|
|
|
if len(opts) == 0 && g.init {
|
|
|
|
return nil
|
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
size := g.opts.PoolSize
|
|
|
|
ttl := g.opts.PoolTTL
|
|
|
|
|
|
|
|
for _, o := range opts {
|
|
|
|
o(&g.opts)
|
|
|
|
}
|
|
|
|
|
|
|
|
// update pool configuration if the options changed
|
|
|
|
if size != g.opts.PoolSize || ttl != g.opts.PoolTTL {
|
|
|
|
g.pool.Lock()
|
|
|
|
g.pool.size = g.opts.PoolSize
|
|
|
|
g.pool.ttl = int64(g.opts.PoolTTL.Seconds())
|
|
|
|
g.pool.Unlock()
|
|
|
|
}
|
|
|
|
|
2021-03-24 15:32:19 +03:00
|
|
|
if err := g.opts.Broker.Init(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := g.opts.Tracer.Init(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := g.opts.Router.Init(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := g.opts.Logger.Init(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := g.opts.Meter.Init(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if err := g.opts.Transport.Init(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
g.init = true
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) Options() client.Options {
|
|
|
|
return g.opts
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) NewMessage(topic string, msg interface{}, opts ...client.MessageOption) client.Message {
|
2019-07-07 14:44:09 +03:00
|
|
|
return newGRPCEvent(topic, msg, g.opts.ContentType, opts...)
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
|
|
|
|
return newGRPCRequest(service, method, req, g.opts.ContentType, reqOpts...)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
|
2020-02-26 21:34:40 +03:00
|
|
|
if req == nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", "req is nil")
|
|
|
|
} else if rsp == nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", "rsp is nil")
|
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
// make a copy of call opts
|
|
|
|
callOpts := g.opts.CallOptions
|
2022-11-16 23:10:27 +03:00
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
for _, opt := range opts {
|
|
|
|
opt(&callOpts)
|
|
|
|
}
|
|
|
|
|
|
|
|
// check if we already have a deadline
|
|
|
|
d, ok := ctx.Deadline()
|
|
|
|
if !ok {
|
|
|
|
// no deadline so we create a new one
|
2019-12-03 10:25:58 +03:00
|
|
|
var cancel context.CancelFunc
|
|
|
|
ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
|
|
|
|
defer cancel()
|
2019-06-03 20:44:43 +03:00
|
|
|
} else {
|
|
|
|
// got a deadline so no need to setup context
|
|
|
|
// but we need to set the timeout we pass along
|
|
|
|
opt := client.WithRequestTimeout(time.Until(d))
|
|
|
|
opt(&callOpts)
|
|
|
|
}
|
|
|
|
|
|
|
|
// should we noop right here?
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
|
|
|
|
// 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)
|
|
|
|
}
|
|
|
|
|
2020-08-18 00:44:45 +03:00
|
|
|
// use the router passed as a call option, or fallback to the rpc clients router
|
|
|
|
if callOpts.Router == nil {
|
|
|
|
callOpts.Router = g.opts.Router
|
|
|
|
}
|
|
|
|
|
|
|
|
if callOpts.Selector == nil {
|
|
|
|
callOpts.Selector = g.opts.Selector
|
|
|
|
}
|
|
|
|
|
|
|
|
// inject proxy address
|
|
|
|
// TODO: don't even bother using Lookup/Select in this case
|
|
|
|
if len(g.opts.Proxy) > 0 {
|
|
|
|
callOpts.Address = []string{g.opts.Proxy}
|
|
|
|
}
|
|
|
|
|
2022-05-01 23:23:15 +03:00
|
|
|
var next selector.Next
|
2020-08-18 00:44:45 +03:00
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
call := func(i int) error {
|
|
|
|
// 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())
|
|
|
|
}
|
|
|
|
|
|
|
|
// only sleep if greater than 0
|
|
|
|
if t.Seconds() > 0 {
|
|
|
|
time.Sleep(t)
|
|
|
|
}
|
|
|
|
|
2022-05-01 23:23:15 +03:00
|
|
|
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
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-08-18 00:44:45 +03:00
|
|
|
// get the next node
|
|
|
|
node := next()
|
2020-07-02 19:26:45 +03:00
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
// make the call
|
2020-07-02 19:26:45 +03:00
|
|
|
err = gcall(ctx, node, req, rsp, callOpts)
|
2020-07-01 19:06:59 +03:00
|
|
|
|
|
|
|
// record the result of the call to inform future routing decisions
|
2020-09-18 18:27:29 +03:00
|
|
|
if verr := g.opts.Selector.Record(node, err); verr != nil {
|
|
|
|
return verr
|
|
|
|
}
|
2020-07-01 19:06:59 +03:00
|
|
|
|
|
|
|
// try and transform the error to a go-micro error
|
2020-03-16 13:33:38 +03:00
|
|
|
if verr, ok := err.(*errors.Error); ok {
|
|
|
|
return verr
|
|
|
|
}
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
ch := make(chan error, callOpts.Retries+1)
|
|
|
|
var gerr error
|
|
|
|
|
|
|
|
for i := 0; i <= callOpts.Retries; i++ {
|
2019-06-27 08:06:53 +03:00
|
|
|
go func(i int) {
|
2019-06-03 20:44:43 +03:00
|
|
|
ch <- call(i)
|
2019-06-27 08:06:53 +03:00
|
|
|
}(i)
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
|
|
|
|
case err := <-ch:
|
|
|
|
// if the call succeeded lets bail early
|
|
|
|
if err == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
retry, rerr := callOpts.Retry(ctx, req, i, err)
|
|
|
|
if rerr != nil {
|
|
|
|
return rerr
|
|
|
|
}
|
|
|
|
|
|
|
|
if !retry {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
gerr = err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return gerr
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
|
|
|
|
// make a copy of call opts
|
|
|
|
callOpts := g.opts.CallOptions
|
|
|
|
for _, opt := range opts {
|
|
|
|
opt(&callOpts)
|
|
|
|
}
|
|
|
|
|
|
|
|
// #200 - streams shouldn't have a request timeout set on the context
|
|
|
|
|
|
|
|
// should we noop right here?
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
|
2020-06-02 19:56:26 +03:00
|
|
|
// 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)
|
|
|
|
}
|
|
|
|
|
2020-08-18 00:44:45 +03:00
|
|
|
// use the router passed as a call option, or fallback to the rpc clients router
|
|
|
|
if callOpts.Router == nil {
|
|
|
|
callOpts.Router = g.opts.Router
|
|
|
|
}
|
|
|
|
|
|
|
|
if callOpts.Selector == nil {
|
|
|
|
callOpts.Selector = g.opts.Selector
|
|
|
|
}
|
|
|
|
|
|
|
|
// inject proxy address
|
|
|
|
// TODO: don't even bother using Lookup/Select in this case
|
|
|
|
if len(g.opts.Proxy) > 0 {
|
|
|
|
callOpts.Address = []string{g.opts.Proxy}
|
|
|
|
}
|
|
|
|
|
2022-05-01 23:23:15 +03:00
|
|
|
var next selector.Next
|
2020-08-18 00:44:45 +03:00
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
call := func(i int) (client.Stream, error) {
|
|
|
|
// 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())
|
|
|
|
}
|
|
|
|
|
|
|
|
// only sleep if greater than 0
|
|
|
|
if t.Seconds() > 0 {
|
|
|
|
time.Sleep(t)
|
|
|
|
}
|
|
|
|
|
2022-05-01 23:23:15 +03:00
|
|
|
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
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-08-18 00:44:45 +03:00
|
|
|
// get the next node
|
|
|
|
node := next()
|
2020-07-02 19:26:45 +03:00
|
|
|
|
2020-06-02 19:56:26 +03:00
|
|
|
// make the call
|
|
|
|
stream := &grpcStream{}
|
2020-11-26 01:45:58 +03:00
|
|
|
err = gstream(ctx, node, req, stream, callOpts)
|
2020-06-02 19:56:26 +03:00
|
|
|
|
2020-07-01 19:06:59 +03:00
|
|
|
// record the result of the call to inform future routing decisions
|
2020-09-18 18:27:29 +03:00
|
|
|
if verr := g.opts.Selector.Record(node, err); verr != nil {
|
|
|
|
return nil, verr
|
|
|
|
}
|
2020-07-01 19:06:59 +03:00
|
|
|
|
|
|
|
// try and transform the error to a go-micro error
|
|
|
|
if verr, ok := err.(*errors.Error); ok {
|
|
|
|
return nil, verr
|
|
|
|
}
|
|
|
|
|
2020-10-28 10:07:30 +03:00
|
|
|
if rerr := g.opts.Selector.Record(node, err); rerr != nil {
|
|
|
|
return nil, rerr
|
|
|
|
}
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
return stream, err
|
|
|
|
}
|
|
|
|
|
|
|
|
type response struct {
|
|
|
|
stream client.Stream
|
|
|
|
err error
|
|
|
|
}
|
|
|
|
|
|
|
|
ch := make(chan response, callOpts.Retries+1)
|
|
|
|
var grr error
|
|
|
|
|
|
|
|
for i := 0; i <= callOpts.Retries; i++ {
|
2019-06-27 08:06:53 +03:00
|
|
|
go func(i int) {
|
2019-06-03 20:44:43 +03:00
|
|
|
s, err := call(i)
|
|
|
|
ch <- response{s, err}
|
2019-06-27 08:06:53 +03:00
|
|
|
}(i)
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
|
|
|
|
case rsp := <-ch:
|
|
|
|
// if the call succeeded lets bail early
|
|
|
|
if rsp.err == nil {
|
|
|
|
return rsp.stream, nil
|
|
|
|
}
|
|
|
|
|
2020-07-01 19:06:59 +03:00
|
|
|
retry, rerr := callOpts.Retry(ctx, req, i, grr)
|
2019-06-03 20:44:43 +03:00
|
|
|
if rerr != nil {
|
|
|
|
return nil, rerr
|
|
|
|
}
|
|
|
|
|
|
|
|
if !retry {
|
|
|
|
return nil, rsp.err
|
|
|
|
}
|
|
|
|
|
|
|
|
grr = rsp.err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil, grr
|
|
|
|
}
|
|
|
|
|
2021-07-31 15:10:59 +03:00
|
|
|
func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
|
|
|
|
return g.publish(ctx, ps, opts...)
|
|
|
|
}
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
2021-07-31 15:10:59 +03:00
|
|
|
return g.publish(ctx, []client.Message{p}, opts...)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
|
2020-07-16 15:35:06 +03:00
|
|
|
var body []byte
|
|
|
|
|
2020-10-28 10:07:30 +03:00
|
|
|
options := client.NewPublishOptions(opts...)
|
2019-11-25 19:31:43 +03:00
|
|
|
|
2021-07-31 15:10:59 +03:00
|
|
|
// get proxy
|
|
|
|
exchange := ""
|
|
|
|
if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
|
|
|
|
exchange = v
|
|
|
|
}
|
|
|
|
|
|
|
|
msgs := make([]*broker.Message, 0, len(ps))
|
|
|
|
|
|
|
|
omd, ok := metadata.FromOutgoingContext(ctx)
|
2019-06-03 20:44:43 +03:00
|
|
|
if !ok {
|
2021-07-31 15:10:59 +03:00
|
|
|
omd = metadata.New(2)
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
2021-07-31 15:10:59 +03:00
|
|
|
for _, p := range ps {
|
|
|
|
md := metadata.Copy(omd)
|
|
|
|
md[metadata.HeaderContentType] = p.ContentType()
|
|
|
|
|
2023-10-26 03:12:32 +03:00
|
|
|
iter := p.Metadata().Iterator()
|
|
|
|
var k, v string
|
|
|
|
for iter.Next(&k, &v) {
|
|
|
|
md.Set(k, v)
|
|
|
|
}
|
|
|
|
|
2021-07-31 15:10:59 +03:00
|
|
|
// passed in raw data
|
|
|
|
if d, ok := p.Payload().(*codec.Frame); ok {
|
|
|
|
body = d.Data
|
|
|
|
} else {
|
|
|
|
// use codec for payload
|
|
|
|
cf, err := g.newCodec(p.ContentType())
|
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
// set the body
|
|
|
|
b, err := cf.Marshal(p.Payload())
|
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
body = b
|
2019-11-25 19:31:43 +03:00
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
|
2021-07-31 15:10:59 +03:00
|
|
|
topic := p.Topic()
|
|
|
|
if len(exchange) > 0 {
|
|
|
|
topic = exchange
|
|
|
|
}
|
2019-11-25 19:31:43 +03:00
|
|
|
|
2022-03-05 19:14:01 +03:00
|
|
|
for k, v := range p.Metadata() {
|
|
|
|
md.Set(k, v)
|
|
|
|
}
|
2021-07-31 15:10:59 +03:00
|
|
|
md.Set(metadata.HeaderTopic, topic)
|
|
|
|
msgs = append(msgs, &broker.Message{Header: md, Body: body})
|
2019-11-25 19:31:43 +03:00
|
|
|
}
|
|
|
|
|
2021-07-31 15:10:59 +03:00
|
|
|
return g.opts.Broker.BatchPublish(ctx, msgs,
|
2021-06-13 12:37:01 +03:00
|
|
|
broker.PublishContext(options.Context),
|
|
|
|
broker.PublishBodyOnly(options.BodyOnly),
|
|
|
|
)
|
2019-06-03 20:44:43 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *grpcClient) String() string {
|
|
|
|
return "grpc"
|
|
|
|
}
|
|
|
|
|
2021-01-29 16:12:28 +03:00
|
|
|
func (g *grpcClient) Name() string {
|
|
|
|
return g.opts.Name
|
|
|
|
}
|
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
func (g *grpcClient) getGrpcDialOptions(ctx context.Context) []grpc.DialOption {
|
|
|
|
var opts []grpc.DialOption
|
2019-10-09 09:45:51 +03:00
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
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...)
|
|
|
|
}
|
|
|
|
}
|
2019-10-09 09:45:51 +03:00
|
|
|
}
|
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
if ctx != nil {
|
|
|
|
if v := ctx.Value(grpcDialOptions{}); v != nil {
|
|
|
|
if vopts, ok := v.([]grpc.DialOption); ok {
|
|
|
|
opts = append(opts, vopts...)
|
|
|
|
}
|
|
|
|
}
|
2019-10-09 09:45:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
return opts
|
|
|
|
}
|
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
func (g *grpcClient) getGrpcCallOptions(ctx context.Context) []grpc.CallOption {
|
|
|
|
var opts []grpc.CallOption
|
2019-10-09 09:45:51 +03:00
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
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...)
|
|
|
|
}
|
|
|
|
}
|
2019-10-09 09:45:51 +03:00
|
|
|
}
|
|
|
|
|
2022-11-16 23:10:27 +03:00
|
|
|
if ctx != nil {
|
|
|
|
if v := ctx.Value(grpcCallOptions{}); v != nil {
|
|
|
|
if vopts, ok := v.([]grpc.CallOption); ok {
|
|
|
|
opts = append(opts, vopts...)
|
|
|
|
}
|
|
|
|
}
|
2019-10-09 09:45:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
return opts
|
|
|
|
}
|
|
|
|
|
2021-01-29 16:12:28 +03:00
|
|
|
func NewClient(opts ...client.Option) client.Client {
|
2020-11-26 01:18:35 +03:00
|
|
|
options := client.NewOptions(opts...)
|
2019-12-30 00:07:55 +03:00
|
|
|
// default content type for grpc
|
2022-11-16 23:10:27 +03:00
|
|
|
if options.ContentType == "" {
|
|
|
|
options.ContentType = DefaultContentType
|
|
|
|
}
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
rc := &grpcClient{
|
|
|
|
opts: options,
|
|
|
|
}
|
2019-12-30 00:07:55 +03:00
|
|
|
|
2019-12-27 15:25:58 +03:00
|
|
|
rc.pool = newPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
|
2019-06-03 20:44:43 +03:00
|
|
|
|
|
|
|
c := client.Client(rc)
|
|
|
|
|
|
|
|
// wrap in reverse
|
|
|
|
for i := len(options.Wrappers); i > 0; i-- {
|
|
|
|
c = options.Wrappers[i-1](c)
|
|
|
|
}
|
|
|
|
|
2020-09-18 18:27:29 +03:00
|
|
|
if rc.opts.Context != nil {
|
2020-11-26 01:18:35 +03:00
|
|
|
if codecs, ok := rc.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
|
|
|
|
for k, v := range codecs {
|
|
|
|
rc.opts.Codecs[k] = &wrapGrpcCodec{v}
|
|
|
|
}
|
2020-09-18 18:27:29 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-11-26 01:18:35 +03:00
|
|
|
for _, k := range options.Codecs {
|
|
|
|
encoding.RegisterCodec(&wrapMicroCodec{k})
|
2020-09-18 18:27:29 +03:00
|
|
|
}
|
|
|
|
|
2019-06-03 20:44:43 +03:00
|
|
|
return c
|
|
|
|
}
|