Compare commits
10 Commits
Author | SHA1 | Date | |
---|---|---|---|
25618a3859 | |||
b22f150601 | |||
71bcb63b60 | |||
698bfbc6f1 | |||
209042de3a | |||
990685628d | |||
0f8ead6acc | |||
2fbcee7b59 | |||
19a469c4e2 | |||
2a6a93a792 |
24
.gitignore
vendored
Normal file
24
.gitignore
vendored
Normal file
@ -0,0 +1,24 @@
|
||||
# Binaries for programs and plugins
|
||||
*.exe
|
||||
*.exe~
|
||||
*.dll
|
||||
*.so
|
||||
*.dylib
|
||||
bin
|
||||
|
||||
# Test binary, built with `go test -c`
|
||||
*.test
|
||||
|
||||
# Output of the go coverage tool, specifically when used with LiteIDE
|
||||
*.out
|
||||
|
||||
# Dependency directories (remove the comment below to include it)
|
||||
# vendor/
|
||||
|
||||
# Go workspace file
|
||||
go.work
|
||||
|
||||
# General
|
||||
.DS_Store
|
||||
.idea
|
||||
.vscode
|
62
codec.go
62
codec.go
@ -1,8 +1,6 @@
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"io"
|
||||
|
||||
"go.unistack.org/micro/v3/codec"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/encoding"
|
||||
@ -65,63 +63,3 @@ func (w *wrapGrpcCodec) Unmarshal(d []byte, v interface{}, opts ...codec.Option)
|
||||
}
|
||||
return w.Codec.Unmarshal(d, v)
|
||||
}
|
||||
|
||||
/*
|
||||
type grpcCodec struct {
|
||||
grpc.ServerStream
|
||||
// headers
|
||||
id string
|
||||
target string
|
||||
method string
|
||||
endpoint string
|
||||
|
||||
c encoding.Codec
|
||||
}
|
||||
|
||||
*/
|
||||
|
||||
func (w *wrapGrpcCodec) ReadHeader(conn io.Reader, m *codec.Message, mt codec.MessageType) error {
|
||||
/*
|
||||
if m == nil {
|
||||
m = codec.NewMessage(codec.Request)
|
||||
}
|
||||
|
||||
if md, ok := metadata.FromIncomingContext(g.ServerStream.Context()); ok {
|
||||
if m.Header == nil {
|
||||
m.Header = meta.New(len(md))
|
||||
}
|
||||
for k, v := range md {
|
||||
m.Header[k] = strings.Join(v, ",")
|
||||
}
|
||||
}
|
||||
|
||||
m.Id = g.id
|
||||
m.Target = g.target
|
||||
m.Method = g.method
|
||||
m.Endpoint = g.endpoint
|
||||
*/
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
|
||||
// caller has requested a frame
|
||||
if m, ok := v.(*codec.Frame); ok {
|
||||
_, err := conn.Read(m.Data)
|
||||
return err
|
||||
}
|
||||
return codec.ErrInvalidMessage
|
||||
}
|
||||
|
||||
func (w *wrapGrpcCodec) Write(conn io.Writer, m *codec.Message, v interface{}) error {
|
||||
// if we don't have a body
|
||||
if v != nil {
|
||||
b, err := w.Marshal(v)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.Body = b
|
||||
}
|
||||
// write the body using the framing codec
|
||||
_, err := conn.Write(m.Body)
|
||||
return err
|
||||
}
|
||||
|
20
go.mod
20
go.mod
@ -1,17 +1,19 @@
|
||||
module go.unistack.org/micro-client-grpc/v3
|
||||
|
||||
go 1.20
|
||||
go 1.21
|
||||
|
||||
toolchain go1.23.1
|
||||
|
||||
require (
|
||||
go.unistack.org/micro/v3 v3.10.28
|
||||
google.golang.org/grpc v1.59.0
|
||||
go.unistack.org/micro/v3 v3.10.91
|
||||
google.golang.org/grpc v1.67.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/golang/protobuf v1.5.3 // indirect
|
||||
golang.org/x/net v0.17.0 // indirect
|
||||
golang.org/x/sys v0.13.0 // indirect
|
||||
golang.org/x/text v0.13.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
|
||||
google.golang.org/protobuf v1.31.0 // indirect
|
||||
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
|
||||
golang.org/x/net v0.29.0 // indirect
|
||||
golang.org/x/sys v0.25.0 // indirect
|
||||
golang.org/x/text v0.18.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
|
||||
google.golang.org/protobuf v1.34.2 // indirect
|
||||
)
|
||||
|
40
go.sum
40
go.sum
@ -1,22 +1,18 @@
|
||||
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
|
||||
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
|
||||
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
||||
go.unistack.org/micro/v3 v3.10.28 h1:/87lGekrmi0/66pioy+Nh8lVUBBYnVqKoHiNYX5OmMI=
|
||||
go.unistack.org/micro/v3 v3.10.28/go.mod h1:eUgtvbtiiz6te93m0ZdmoecbitWwjdBmmr84srmEIKA=
|
||||
golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
|
||||
golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
|
||||
golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
|
||||
golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
|
||||
golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b h1:ZlWIi1wSK56/8hn4QcBp/j9M7Gt3U/3hZw3mC7vDICo=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b/go.mod h1:swOH3j0KzcDDgGUWr+SNpyTen5YrXjS3eyPzFYKc6lc=
|
||||
google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
|
||||
google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
|
||||
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
|
||||
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
|
||||
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
|
||||
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
|
||||
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
||||
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
|
||||
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
|
||||
go.unistack.org/micro/v3 v3.10.91 h1:vuJY4tXwpqimwIkEJ3TozMYNVQQs+C5QMlQWPgSY/YM=
|
||||
go.unistack.org/micro/v3 v3.10.91/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
|
||||
golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
|
||||
golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
|
||||
golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
|
||||
golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
|
||||
golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
|
||||
google.golang.org/grpc v1.67.0 h1:IdH9y6PF5MPSdAntIcpjQ+tXO41pcQsfZV2RxtQgVcw=
|
||||
google.golang.org/grpc v1.67.0/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA=
|
||||
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
|
||||
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
|
||||
|
160
grpc.go
160
grpc.go
@ -1,5 +1,5 @@
|
||||
// Package grpc provides a gRPC client
|
||||
package grpc // import "go.unistack.org/micro-client-grpc/v3"
|
||||
// Package grpc provides a gRPC client for micro framework
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
@ -8,6 +8,7 @@ import (
|
||||
"net"
|
||||
"os"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@ -17,12 +18,17 @@ import (
|
||||
"go.unistack.org/micro/v3/codec"
|
||||
"go.unistack.org/micro/v3/errors"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
"go.unistack.org/micro/v3/options"
|
||||
"go.unistack.org/micro/v3/selector"
|
||||
"go.unistack.org/micro/v3/semconv"
|
||||
"go.unistack.org/micro/v3/tracer"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/encoding"
|
||||
gmetadata "google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const (
|
||||
@ -30,8 +36,12 @@ const (
|
||||
)
|
||||
|
||||
type grpcClient struct {
|
||||
pool *ConnPool
|
||||
opts client.Options
|
||||
funcPublish client.FuncPublish
|
||||
funcBatchPublish client.FuncBatchPublish
|
||||
funcCall client.FuncCall
|
||||
funcStream client.FuncStream
|
||||
pool *ConnPool
|
||||
opts client.Options
|
||||
sync.RWMutex
|
||||
init bool
|
||||
}
|
||||
@ -98,6 +108,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
|
||||
|
||||
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
||||
maxSendMsgSize := g.maxSendMsgSizeValue()
|
||||
cfgService := g.serviceConfig()
|
||||
|
||||
var grr error
|
||||
|
||||
@ -116,6 +127,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
|
||||
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
||||
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
||||
),
|
||||
grpc.WithDefaultServiceConfig(cfgService),
|
||||
}
|
||||
|
||||
if opts := g.getGrpcDialOptions(g.opts.Context); opts != nil {
|
||||
@ -221,6 +233,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
|
||||
|
||||
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
||||
maxSendMsgSize := g.maxSendMsgSizeValue()
|
||||
cfgService := g.serviceConfig()
|
||||
|
||||
grpcDialOptions := []grpc.DialOption{
|
||||
g.secure(addr),
|
||||
@ -228,6 +241,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
|
||||
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
||||
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
||||
),
|
||||
grpc.WithDefaultServiceConfig(cfgService),
|
||||
}
|
||||
|
||||
if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
|
||||
@ -372,6 +386,17 @@ func (g *grpcClient) newCodec(ct string) (codec.Codec, error) {
|
||||
return nil, codec.ErrUnknownContentType
|
||||
}
|
||||
|
||||
func (g *grpcClient) serviceConfig() string {
|
||||
if g.opts.Context == nil {
|
||||
return DefaultServiceConfig
|
||||
}
|
||||
v := g.opts.Context.Value(serviceConfigKey{})
|
||||
if v == nil {
|
||||
return DefaultServiceConfig
|
||||
}
|
||||
return v.(string)
|
||||
}
|
||||
|
||||
func (g *grpcClient) Init(opts ...client.Option) error {
|
||||
if len(opts) == 0 && g.init {
|
||||
return nil
|
||||
@ -391,26 +416,23 @@ func (g *grpcClient) Init(opts ...client.Option) error {
|
||||
g.pool.Unlock()
|
||||
}
|
||||
|
||||
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.funcCall = g.fnCall
|
||||
g.funcStream = g.fnStream
|
||||
g.funcPublish = g.fnPublish
|
||||
g.funcBatchPublish = g.fnBatchPublish
|
||||
|
||||
g.init = true
|
||||
g.opts.Hooks.EachNext(func(hook options.Hook) {
|
||||
switch h := hook.(type) {
|
||||
case client.HookCall:
|
||||
g.funcCall = h(g.funcCall)
|
||||
case client.HookStream:
|
||||
g.funcStream = h(g.funcStream)
|
||||
case client.HookPublish:
|
||||
g.funcPublish = h(g.funcPublish)
|
||||
case client.HookBatchPublish:
|
||||
g.funcBatchPublish = h(g.funcBatchPublish)
|
||||
}
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -433,6 +455,32 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
|
||||
} else if rsp == nil {
|
||||
return errors.InternalServerError("go.micro.client", "rsp is nil")
|
||||
}
|
||||
|
||||
ts := time.Now()
|
||||
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
|
||||
var sp tracer.Span
|
||||
ctx, sp = g.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
|
||||
tracer.WithSpanKind(tracer.SpanKindClient),
|
||||
tracer.WithSpanLabels("endpoint", req.Endpoint()),
|
||||
)
|
||||
err := g.funcCall(ctx, req, rsp, opts...)
|
||||
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
|
||||
te := time.Since(ts)
|
||||
g.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
|
||||
g.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
|
||||
|
||||
if me := errors.FromError(err); me == nil {
|
||||
sp.Finish()
|
||||
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(200))).Inc()
|
||||
} else {
|
||||
sp.SetStatus(tracer.SpanStatusError, err.Error())
|
||||
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code))).Inc()
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
|
||||
// make a copy of call opts
|
||||
callOpts := g.opts.CallOptions
|
||||
|
||||
@ -464,11 +512,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
|
||||
@ -568,6 +611,31 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
|
||||
}
|
||||
|
||||
func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
|
||||
ts := time.Now()
|
||||
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
|
||||
var sp tracer.Span
|
||||
ctx, sp = g.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
|
||||
tracer.WithSpanKind(tracer.SpanKindClient),
|
||||
tracer.WithSpanLabels("endpoint", req.Endpoint()),
|
||||
)
|
||||
stream, err := g.funcStream(ctx, req, opts...)
|
||||
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
|
||||
te := time.Since(ts)
|
||||
g.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
|
||||
g.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
|
||||
|
||||
if me := status.Convert(err); me == nil {
|
||||
sp.Finish()
|
||||
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(codes.OK))).Inc()
|
||||
} else {
|
||||
sp.SetStatus(tracer.SpanStatusError, err.Error())
|
||||
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code()))).Inc()
|
||||
}
|
||||
|
||||
return stream, err
|
||||
}
|
||||
|
||||
func (g *grpcClient) fnStream(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 {
|
||||
@ -586,11 +654,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
|
||||
@ -701,10 +764,18 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
|
||||
}
|
||||
|
||||
func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
|
||||
return g.funcBatchPublish(ctx, ps, opts...)
|
||||
}
|
||||
|
||||
func (g *grpcClient) fnBatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
|
||||
return g.publish(ctx, ps, opts...)
|
||||
}
|
||||
|
||||
func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
||||
return g.funcPublish(ctx, p, opts...)
|
||||
}
|
||||
|
||||
func (g *grpcClient) fnPublish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
||||
return g.publish(ctx, []client.Message{p}, opts...)
|
||||
}
|
||||
|
||||
@ -760,7 +831,6 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
|
||||
}
|
||||
body = b
|
||||
}
|
||||
|
||||
msgs = append(msgs, &broker.Message{Header: md, Body: body})
|
||||
}
|
||||
|
||||
@ -829,23 +899,16 @@ func NewClient(opts ...client.Option) client.Client {
|
||||
options.ContentType = DefaultContentType
|
||||
}
|
||||
|
||||
rc := &grpcClient{
|
||||
c := &grpcClient{
|
||||
opts: options,
|
||||
}
|
||||
|
||||
rc.pool = NewConnPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
|
||||
c.pool = NewConnPool(options.PoolSize, options.PoolTTL, c.poolMaxIdle(), c.poolMaxStreams())
|
||||
|
||||
c := client.Client(rc)
|
||||
|
||||
// wrap in reverse
|
||||
for i := len(options.Wrappers); i > 0; i-- {
|
||||
c = options.Wrappers[i-1](c)
|
||||
}
|
||||
|
||||
if rc.opts.Context != nil {
|
||||
if codecs, ok := rc.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
|
||||
if c.opts.Context != nil {
|
||||
if codecs, ok := c.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
|
||||
for k, v := range codecs {
|
||||
rc.opts.Codecs[k] = &wrapGrpcCodec{v}
|
||||
c.opts.Codecs[k] = &wrapGrpcCodec{v}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -854,5 +917,10 @@ func NewClient(opts ...client.Option) client.Client {
|
||||
encoding.RegisterCodec(&wrapMicroCodec{k})
|
||||
}
|
||||
|
||||
c.funcCall = c.fnCall
|
||||
c.funcStream = c.fnStream
|
||||
c.funcPublish = c.fnPublish
|
||||
c.funcBatchPublish = c.fnBatchPublish
|
||||
|
||||
return c
|
||||
}
|
||||
|
24
grpc_pool.go
24
grpc_pool.go
@ -21,22 +21,22 @@ type ConnPool struct {
|
||||
|
||||
type streamsPool struct {
|
||||
// head of list
|
||||
head *poolConn
|
||||
head *PoolConn
|
||||
// busy conns list
|
||||
busy *poolConn
|
||||
busy *PoolConn
|
||||
// the siza of list
|
||||
count int
|
||||
// idle conn
|
||||
idle int
|
||||
}
|
||||
|
||||
type poolConn struct {
|
||||
type PoolConn struct {
|
||||
err error
|
||||
*grpc.ClientConn
|
||||
next *poolConn
|
||||
next *PoolConn
|
||||
pool *ConnPool
|
||||
sp *streamsPool
|
||||
pre *poolConn
|
||||
pre *PoolConn
|
||||
addr string
|
||||
streams int
|
||||
created int64
|
||||
@ -59,7 +59,7 @@ func NewConnPool(size int, ttl time.Duration, idle int, ms int) *ConnPool {
|
||||
}
|
||||
}
|
||||
|
||||
func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption) (*poolConn, error) {
|
||||
func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption) (*PoolConn, error) {
|
||||
if strings.HasPrefix(addr, "http") {
|
||||
addr = addr[strings.Index(addr, ":")+3:]
|
||||
}
|
||||
@ -67,7 +67,7 @@ func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption
|
||||
p.Lock()
|
||||
sp, ok := p.conns[addr]
|
||||
if !ok {
|
||||
sp = &streamsPool{head: &poolConn{}, busy: &poolConn{}, count: 0, idle: 0}
|
||||
sp = &streamsPool{head: &PoolConn{}, busy: &PoolConn{}, count: 0, idle: 0}
|
||||
p.conns[addr] = sp
|
||||
}
|
||||
// while we have conns check streams and then return one
|
||||
@ -135,7 +135,7 @@ func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conn = &poolConn{ClientConn: cc, err: nil, addr: addr, pool: p, sp: sp, streams: 1, created: time.Now().Unix(), pre: nil, next: nil, in: false}
|
||||
conn = &PoolConn{ClientConn: cc, err: nil, addr: addr, pool: p, sp: sp, streams: 1, created: time.Now().Unix(), pre: nil, next: nil, in: false}
|
||||
|
||||
// add conn to streams pool
|
||||
p.Lock()
|
||||
@ -147,7 +147,7 @@ func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (p *ConnPool) Put(conn *poolConn, err error) {
|
||||
func (p *ConnPool) Put(conn *PoolConn, err error) {
|
||||
p.Lock()
|
||||
p, sp, created := conn.pool, conn.sp, conn.created
|
||||
// try to add conn
|
||||
@ -182,11 +182,11 @@ func (p *ConnPool) Put(conn *poolConn, err error) {
|
||||
p.Unlock()
|
||||
}
|
||||
|
||||
func (conn *poolConn) Close() {
|
||||
func (conn *PoolConn) Close() {
|
||||
conn.pool.Put(conn, conn.err)
|
||||
}
|
||||
|
||||
func removeConn(conn *poolConn) {
|
||||
func removeConn(conn *PoolConn) {
|
||||
if conn.pre != nil {
|
||||
conn.pre.next = conn.next
|
||||
}
|
||||
@ -199,7 +199,7 @@ func removeConn(conn *poolConn) {
|
||||
conn.sp.count--
|
||||
}
|
||||
|
||||
func addConnAfter(conn *poolConn, after *poolConn) {
|
||||
func addConnAfter(conn *PoolConn, after *PoolConn) {
|
||||
conn.next = after.next
|
||||
conn.pre = after
|
||||
if after.next != nil {
|
||||
|
16
options.go
16
options.go
@ -10,7 +10,7 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
// DefaultPoolMaxStreams maximum streams on a connectioin
|
||||
// DefaultPoolMaxStreams maximum streams on a connection
|
||||
// (20)
|
||||
DefaultPoolMaxStreams = 20
|
||||
|
||||
@ -25,6 +25,9 @@ var (
|
||||
// DefaultMaxSendMsgSize maximum message that client can send
|
||||
// (4 MB).
|
||||
DefaultMaxSendMsgSize = 1024 * 1024 * 4
|
||||
|
||||
// DefaultServiceConfig enable load balancing
|
||||
DefaultServiceConfig = `{"loadBalancingPolicy":"round_robin"}`
|
||||
)
|
||||
|
||||
type poolMaxStreams struct{}
|
||||
@ -115,3 +118,14 @@ func CallOptions(opts ...grpc.CallOption) client.CallOption {
|
||||
o.Context = context.WithValue(o.Context, grpcCallOptions{}, opts)
|
||||
}
|
||||
}
|
||||
|
||||
type serviceConfigKey struct{}
|
||||
|
||||
func ServiceConfig(str string) client.CallOption {
|
||||
return func(options *client.CallOptions) {
|
||||
if options.Context == nil {
|
||||
options.Context = context.Background()
|
||||
}
|
||||
options.Context = context.WithValue(options.Context, serviceConfigKey{}, str)
|
||||
}
|
||||
}
|
||||
|
@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
type response struct {
|
||||
conn *poolConn
|
||||
conn *PoolConn
|
||||
stream grpc.ClientStream
|
||||
codec codec.Codec
|
||||
}
|
||||
@ -34,9 +34,5 @@ func (r *response) Header() metadata.Metadata {
|
||||
|
||||
// Read the undecoded response
|
||||
func (r *response) Read() ([]byte, error) {
|
||||
f := &codec.Frame{}
|
||||
if err := r.codec.ReadBody(&wrapStream{r.stream}, f); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return f.Data, nil
|
||||
return nil, nil
|
||||
}
|
||||
|
@ -6,6 +6,7 @@ import (
|
||||
"sync"
|
||||
|
||||
"go.unistack.org/micro/v3/client"
|
||||
"go.unistack.org/micro/v3/tracer"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
@ -17,7 +18,7 @@ type grpcStream struct {
|
||||
request client.Request
|
||||
response client.Response
|
||||
close func(err error)
|
||||
conn *poolConn
|
||||
conn *PoolConn
|
||||
sync.RWMutex
|
||||
closed bool
|
||||
}
|
||||
@ -111,6 +112,12 @@ func (g *grpcStream) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
if sp, ok := tracer.SpanFromContext(g.context); ok && sp != nil {
|
||||
if g.err != nil {
|
||||
sp.SetStatus(tracer.SpanStatusError, g.err.Error())
|
||||
}
|
||||
sp.Finish()
|
||||
}
|
||||
// close the connection
|
||||
g.closed = true
|
||||
g.close(g.err)
|
||||
|
Loading…
Reference in New Issue
Block a user