Compare commits
9 Commits
Author | SHA1 | Date | |
---|---|---|---|
25618a3859 | |||
b22f150601 | |||
71bcb63b60 | |||
698bfbc6f1 | |||
209042de3a | |||
990685628d | |||
0f8ead6acc | |||
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
|
package grpc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"io"
|
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/codec"
|
"go.unistack.org/micro/v3/codec"
|
||||||
"google.golang.org/grpc"
|
"google.golang.org/grpc"
|
||||||
"google.golang.org/grpc/encoding"
|
"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)
|
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
|
module go.unistack.org/micro-client-grpc/v3
|
||||||
|
|
||||||
go 1.20
|
go 1.21
|
||||||
|
|
||||||
|
toolchain go1.23.1
|
||||||
|
|
||||||
require (
|
require (
|
||||||
go.unistack.org/micro/v3 v3.10.28
|
go.unistack.org/micro/v3 v3.10.91
|
||||||
google.golang.org/grpc v1.59.0
|
google.golang.org/grpc v1.67.0
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/golang/protobuf v1.5.3 // indirect
|
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
|
||||||
golang.org/x/net v0.17.0 // indirect
|
golang.org/x/net v0.29.0 // indirect
|
||||||
golang.org/x/sys v0.13.0 // indirect
|
golang.org/x/sys v0.25.0 // indirect
|
||||||
golang.org/x/text v0.13.0 // indirect
|
golang.org/x/text v0.18.0 // indirect
|
||||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
|
||||||
google.golang.org/protobuf v1.31.0 // 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/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
||||||
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
|
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||||
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
|
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
|
||||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
|
||||||
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
go.unistack.org/micro/v3 v3.10.91 h1:vuJY4tXwpqimwIkEJ3TozMYNVQQs+C5QMlQWPgSY/YM=
|
||||||
go.unistack.org/micro/v3 v3.10.28 h1:/87lGekrmi0/66pioy+Nh8lVUBBYnVqKoHiNYX5OmMI=
|
go.unistack.org/micro/v3 v3.10.91/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
|
||||||
go.unistack.org/micro/v3 v3.10.28/go.mod h1:eUgtvbtiiz6te93m0ZdmoecbitWwjdBmmr84srmEIKA=
|
golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
|
||||||
golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
|
golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
|
||||||
golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
|
golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
|
||||||
golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
|
golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||||
golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
|
||||||
golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
|
golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
|
||||||
golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
|
||||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
|
||||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b h1:ZlWIi1wSK56/8hn4QcBp/j9M7Gt3U/3hZw3mC7vDICo=
|
google.golang.org/grpc v1.67.0 h1:IdH9y6PF5MPSdAntIcpjQ+tXO41pcQsfZV2RxtQgVcw=
|
||||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b/go.mod h1:swOH3j0KzcDDgGUWr+SNpyTen5YrXjS3eyPzFYKc6lc=
|
google.golang.org/grpc v1.67.0/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA=
|
||||||
google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
|
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
|
||||||
google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
|
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
|
||||||
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=
|
|
||||||
|
156
grpc.go
156
grpc.go
@ -1,5 +1,5 @@
|
|||||||
// Package grpc provides a gRPC client
|
// Package grpc provides a gRPC client for micro framework
|
||||||
package grpc // import "go.unistack.org/micro-client-grpc/v3"
|
package grpc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
@ -8,6 +8,7 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"os"
|
"os"
|
||||||
"reflect"
|
"reflect"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
@ -17,12 +18,17 @@ import (
|
|||||||
"go.unistack.org/micro/v3/codec"
|
"go.unistack.org/micro/v3/codec"
|
||||||
"go.unistack.org/micro/v3/errors"
|
"go.unistack.org/micro/v3/errors"
|
||||||
"go.unistack.org/micro/v3/metadata"
|
"go.unistack.org/micro/v3/metadata"
|
||||||
|
"go.unistack.org/micro/v3/options"
|
||||||
"go.unistack.org/micro/v3/selector"
|
"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"
|
||||||
|
"google.golang.org/grpc/codes"
|
||||||
"google.golang.org/grpc/credentials"
|
"google.golang.org/grpc/credentials"
|
||||||
"google.golang.org/grpc/credentials/insecure"
|
"google.golang.org/grpc/credentials/insecure"
|
||||||
"google.golang.org/grpc/encoding"
|
"google.golang.org/grpc/encoding"
|
||||||
gmetadata "google.golang.org/grpc/metadata"
|
gmetadata "google.golang.org/grpc/metadata"
|
||||||
|
"google.golang.org/grpc/status"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@ -30,6 +36,10 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type grpcClient struct {
|
type grpcClient struct {
|
||||||
|
funcPublish client.FuncPublish
|
||||||
|
funcBatchPublish client.FuncBatchPublish
|
||||||
|
funcCall client.FuncCall
|
||||||
|
funcStream client.FuncStream
|
||||||
pool *ConnPool
|
pool *ConnPool
|
||||||
opts client.Options
|
opts client.Options
|
||||||
sync.RWMutex
|
sync.RWMutex
|
||||||
@ -98,6 +108,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
|
|||||||
|
|
||||||
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
||||||
maxSendMsgSize := g.maxSendMsgSizeValue()
|
maxSendMsgSize := g.maxSendMsgSizeValue()
|
||||||
|
cfgService := g.serviceConfig()
|
||||||
|
|
||||||
var grr error
|
var grr error
|
||||||
|
|
||||||
@ -116,6 +127,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
|
|||||||
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
||||||
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
||||||
),
|
),
|
||||||
|
grpc.WithDefaultServiceConfig(cfgService),
|
||||||
}
|
}
|
||||||
|
|
||||||
if opts := g.getGrpcDialOptions(g.opts.Context); opts != nil {
|
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()
|
maxRecvMsgSize := g.maxRecvMsgSizeValue()
|
||||||
maxSendMsgSize := g.maxSendMsgSizeValue()
|
maxSendMsgSize := g.maxSendMsgSizeValue()
|
||||||
|
cfgService := g.serviceConfig()
|
||||||
|
|
||||||
grpcDialOptions := []grpc.DialOption{
|
grpcDialOptions := []grpc.DialOption{
|
||||||
g.secure(addr),
|
g.secure(addr),
|
||||||
@ -228,6 +241,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
|
|||||||
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
|
||||||
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
grpc.MaxCallSendMsgSize(maxSendMsgSize),
|
||||||
),
|
),
|
||||||
|
grpc.WithDefaultServiceConfig(cfgService),
|
||||||
}
|
}
|
||||||
|
|
||||||
if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
|
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
|
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 {
|
func (g *grpcClient) Init(opts ...client.Option) error {
|
||||||
if len(opts) == 0 && g.init {
|
if len(opts) == 0 && g.init {
|
||||||
return nil
|
return nil
|
||||||
@ -391,26 +416,23 @@ func (g *grpcClient) Init(opts ...client.Option) error {
|
|||||||
g.pool.Unlock()
|
g.pool.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := g.opts.Broker.Init(); err != nil {
|
g.funcCall = g.fnCall
|
||||||
return err
|
g.funcStream = g.fnStream
|
||||||
}
|
g.funcPublish = g.fnPublish
|
||||||
if err := g.opts.Tracer.Init(); err != nil {
|
g.funcBatchPublish = g.fnBatchPublish
|
||||||
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
|
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
|
return nil
|
||||||
}
|
}
|
||||||
@ -433,6 +455,32 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
|
|||||||
} else if rsp == nil {
|
} else if rsp == nil {
|
||||||
return errors.InternalServerError("go.micro.client", "rsp is 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
|
// make a copy of call opts
|
||||||
callOpts := g.opts.CallOptions
|
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
|
// make copy of call method
|
||||||
gcall := g.call
|
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
|
// use the router passed as a call option, or fallback to the rpc clients router
|
||||||
if callOpts.Router == nil {
|
if callOpts.Router == nil {
|
||||||
callOpts.Router = g.opts.Router
|
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) {
|
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
|
// make a copy of call opts
|
||||||
callOpts := g.opts.CallOptions
|
callOpts := g.opts.CallOptions
|
||||||
for _, opt := range opts {
|
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
|
// make a copy of stream
|
||||||
gstream := g.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
|
// use the router passed as a call option, or fallback to the rpc clients router
|
||||||
if callOpts.Router == nil {
|
if callOpts.Router == nil {
|
||||||
callOpts.Router = g.opts.Router
|
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 {
|
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...)
|
return g.publish(ctx, ps, opts...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
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...)
|
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
|
body = b
|
||||||
}
|
}
|
||||||
|
|
||||||
msgs = append(msgs, &broker.Message{Header: md, Body: body})
|
msgs = append(msgs, &broker.Message{Header: md, Body: body})
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -829,23 +899,16 @@ func NewClient(opts ...client.Option) client.Client {
|
|||||||
options.ContentType = DefaultContentType
|
options.ContentType = DefaultContentType
|
||||||
}
|
}
|
||||||
|
|
||||||
rc := &grpcClient{
|
c := &grpcClient{
|
||||||
opts: options,
|
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)
|
if c.opts.Context != nil {
|
||||||
|
if codecs, ok := c.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
|
||||||
// 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 {
|
|
||||||
for k, v := range codecs {
|
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})
|
encoding.RegisterCodec(&wrapMicroCodec{k})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
c.funcCall = c.fnCall
|
||||||
|
c.funcStream = c.fnStream
|
||||||
|
c.funcPublish = c.fnPublish
|
||||||
|
c.funcBatchPublish = c.fnBatchPublish
|
||||||
|
|
||||||
return c
|
return c
|
||||||
}
|
}
|
||||||
|
14
options.go
14
options.go
@ -25,6 +25,9 @@ var (
|
|||||||
// DefaultMaxSendMsgSize maximum message that client can send
|
// DefaultMaxSendMsgSize maximum message that client can send
|
||||||
// (4 MB).
|
// (4 MB).
|
||||||
DefaultMaxSendMsgSize = 1024 * 1024 * 4
|
DefaultMaxSendMsgSize = 1024 * 1024 * 4
|
||||||
|
|
||||||
|
// DefaultServiceConfig enable load balancing
|
||||||
|
DefaultServiceConfig = `{"loadBalancingPolicy":"round_robin"}`
|
||||||
)
|
)
|
||||||
|
|
||||||
type poolMaxStreams struct{}
|
type poolMaxStreams struct{}
|
||||||
@ -115,3 +118,14 @@ func CallOptions(opts ...grpc.CallOption) client.CallOption {
|
|||||||
o.Context = context.WithValue(o.Context, grpcCallOptions{}, opts)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -34,9 +34,5 @@ func (r *response) Header() metadata.Metadata {
|
|||||||
|
|
||||||
// Read the undecoded response
|
// Read the undecoded response
|
||||||
func (r *response) Read() ([]byte, error) {
|
func (r *response) Read() ([]byte, error) {
|
||||||
f := &codec.Frame{}
|
return nil, nil
|
||||||
if err := r.codec.ReadBody(&wrapStream{r.stream}, f); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return f.Data, nil
|
|
||||||
}
|
}
|
||||||
|
@ -6,6 +6,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"go.unistack.org/micro/v3/client"
|
"go.unistack.org/micro/v3/client"
|
||||||
|
"go.unistack.org/micro/v3/tracer"
|
||||||
"google.golang.org/grpc"
|
"google.golang.org/grpc"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -111,6 +112,12 @@ func (g *grpcStream) Close() error {
|
|||||||
return nil
|
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
|
// close the connection
|
||||||
g.closed = true
|
g.closed = true
|
||||||
g.close(g.err)
|
g.close(g.err)
|
||||||
|
Loading…
Reference in New Issue
Block a user