Compare commits

...

13 Commits

Author SHA1 Message Date
d536140a5b allow to publish body only
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-06-13 12:37:01 +03:00
db02559a00 limit send/recv msg size for streams too
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-06-05 16:24:29 +03:00
0fae2e1bdd move default content type to const
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-05-24 23:11:10 +03:00
Renovate Bot
7a1fdef33a Update module google.golang.org/grpc to v1.37.1 2021-05-11 23:44:05 +00:00
2bb81ff232 merge https://github.com/asim/go-micro/pull/2158
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-04-29 14:31:57 +03:00
Renovate Bot
9ac268b2f0 Update module github.com/unistack-org/micro/v3 to v3.3.17 2021-04-27 08:54:25 +00:00
510fa4b379 lint (#32)
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-04-26 19:17:29 +03:00
Renovate Bot
08def4d244 Update module github.com/unistack-org/micro/v3 to v3.3.16 2021-04-20 11:52:53 +00:00
a4683c0b78 drop AuthTLS option and use client TLSConfig option
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-04-20 12:21:40 +03:00
Renovate Bot
637e3df24a Update module github.com/unistack-org/micro/v3 to v3.3.15 2021-04-19 14:55:53 +00:00
Renovate Bot
36c7fa6a23 Update module github.com/unistack-org/micro/v3 to v3.3.14 2021-04-19 04:53:53 +00:00
2bef21a001 minor content-type fixes
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-04-16 17:07:12 +03:00
Renovate Bot
78c32dc3e8 Update module github.com/unistack-org/micro/v3 to v3.3.13 2021-04-10 01:39:46 +00:00
12 changed files with 138 additions and 138 deletions

13
.github/stale.sh vendored
View File

@@ -1,13 +0,0 @@
#!/bin/bash -ex
export PATH=$PATH:$(pwd)/bin
export GO111MODULE=on
export GOBIN=$(pwd)/bin
#go get github.com/rvflash/goup@v0.4.1
#goup -v ./...
#go get github.com/psampaz/go-mod-outdated@v0.6.0
go list -u -m -mod=mod -json all | go-mod-outdated -update -direct -ci || true
#go list -u -m -json all | go-mod-outdated -update

44
.golangci.yml Normal file
View File

@@ -0,0 +1,44 @@
run:
concurrency: 4
deadline: 5m
issues-exit-code: 1
tests: true
linters-settings:
govet:
check-shadowing: true
enable:
- fieldalignment
linters:
enable:
- govet
- deadcode
- errcheck
- govet
- ineffassign
- staticcheck
- structcheck
- typecheck
- unused
- varcheck
- bodyclose
- gci
- goconst
- gocritic
- gosimple
- gofmt
- gofumpt
- goimports
- golint
- gosec
- makezero
- misspell
- nakedret
- nestif
- nilerr
- noctx
- prealloc
- unconvert
- unparam
disable-all: false

View File

@@ -36,19 +36,17 @@ func (w *wrapGrpcCodec) String() string {
}
func (w *wrapGrpcCodec) Marshal(v interface{}) ([]byte, error) {
switch m := v.(type) {
case *codec.Frame:
if m, ok := v.(*codec.Frame); ok {
return m.Data, nil
}
return w.Codec.Marshal(v)
}
func (w wrapGrpcCodec) Unmarshal(d []byte, v interface{}) error {
func (w *wrapGrpcCodec) Unmarshal(d []byte, v interface{}) error {
if d == nil || v == nil {
return nil
}
switch m := v.(type) {
case *codec.Frame:
if m, ok := v.(*codec.Frame); ok {
m.Data = d
return nil
}
@@ -69,7 +67,7 @@ type grpcCodec struct {
*/
func (g *wrapGrpcCodec) ReadHeader(conn io.Reader, m *codec.Message, mt codec.MessageType) error {
func (w *wrapGrpcCodec) ReadHeader(conn io.Reader, m *codec.Message, mt codec.MessageType) error {
/*
if m == nil {
m = codec.NewMessage(codec.Request)
@@ -92,20 +90,19 @@ func (g *wrapGrpcCodec) ReadHeader(conn io.Reader, m *codec.Message, mt codec.Me
return nil
}
func (g *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
func (w *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
// caller has requested a frame
switch m := v.(type) {
case *codec.Frame:
if m, ok := v.(*codec.Frame); ok {
_, err := conn.Read(m.Data)
return err
}
return codec.ErrInvalidMessage
}
func (g *wrapGrpcCodec) Write(conn io.Writer, m *codec.Message, v interface{}) error {
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 := g.Marshal(v)
b, err := w.Marshal(v)
if err != nil {
return err
}

View File

@@ -35,5 +35,10 @@ func microError(err error) error {
}
// fallback
return errors.InternalServerError("go.micro.client", s.Message())
return &errors.Error{
Id: "go.micro.client",
Code: int32(s.Code()),
Detail: s.Message(),
Status: s.Code().String(),
}
}

6
go.mod
View File

@@ -4,10 +4,8 @@ go 1.16
require (
github.com/google/go-cmp v0.5.1 // indirect
github.com/unistack-org/micro/v3 v3.3.11
github.com/unistack-org/micro/v3 v3.3.20
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
google.golang.org/genproto v0.0.0-20200904004341-0bd0a958aa1d // indirect
google.golang.org/grpc v1.37.0
google.golang.org/grpc v1.37.1
)
//replace github.com/unistack-org/micro/v3 => ../../micro

47
go.sum
View File

@@ -8,7 +8,6 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZm
github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
@@ -40,21 +39,8 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:
github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/unistack-org/micro/v3 v3.3.0 h1:pEj/8QVFzMlNMEL//q/Te8qgG+XI6LTYIQrb6hMymgk=
github.com/unistack-org/micro/v3 v3.3.0/go.mod h1:iJwCWq2PECMxigfqe6TPC5GLWvj6P94Kk+PTVZGL3w8=
github.com/unistack-org/micro/v3 v3.3.1 h1:2KUgb5THy5GUjYgMcOf3TXYIJ0h2RbJHb2j6eeWdRx0=
github.com/unistack-org/micro/v3 v3.3.1/go.mod h1:iJwCWq2PECMxigfqe6TPC5GLWvj6P94Kk+PTVZGL3w8=
github.com/unistack-org/micro/v3 v3.3.2 h1:1r7fmaobJVZBKDhIewSBy8R4H/6YazQFetsDgULV6Mw=
github.com/unistack-org/micro/v3 v3.3.2/go.mod h1:tX95c0Qx4w6oqU7qKThs9lya9P507BdZ29MsTVDmU6w=
github.com/unistack-org/micro/v3 v3.3.4 h1:wSYjL1sfrqtMmgFfYKylrKpAepMHrPcOpe10hvQAKuo=
github.com/unistack-org/micro/v3 v3.3.4/go.mod h1:tX95c0Qx4w6oqU7qKThs9lya9P507BdZ29MsTVDmU6w=
github.com/unistack-org/micro/v3 v3.3.8/go.mod h1:M3RGgKZX1OlTn0QB+3R1tlSoGwmEg6rb/Isr0lLYmjQ=
github.com/unistack-org/micro/v3 v3.3.9 h1:NFF4wymDDxgYeZU++04BxvETrrAnd0iNzVyfm/ssQI4=
github.com/unistack-org/micro/v3 v3.3.9/go.mod h1:M3RGgKZX1OlTn0QB+3R1tlSoGwmEg6rb/Isr0lLYmjQ=
github.com/unistack-org/micro/v3 v3.3.10 h1:yMSiyplupFQ7xy/kNXL/zqSnqGSjrlNa2GTCwea5BVg=
github.com/unistack-org/micro/v3 v3.3.10/go.mod h1:5ragE2E8ER5d4FZQJG9pB6qdfOoLXLfKW89l77Dy3jQ=
github.com/unistack-org/micro/v3 v3.3.11 h1:Jr0gAw5lLqgddiHKQeWUOUeP6ZqgRhz52EA9zJ5MJ3U=
github.com/unistack-org/micro/v3 v3.3.11/go.mod h1:PPCt675o3HPcODFbJ4iRWPmQFAk1WQ+asQSOb/syq6U=
github.com/unistack-org/micro/v3 v3.3.20 h1:gB+sPtvYuEKJQG/k5xnC1TK7MnZbr7wlgyqpYNREdyo=
github.com/unistack-org/micro/v3 v3.3.20/go.mod h1:LXmPfbJnJNvL0kQs8HfnkV3Wya2Wb+C7keVq++RCZnk=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
@@ -64,14 +50,8 @@ golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73r
golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4 h1:b0LrWgu8+q7z4J+0Y3Umo5q1dL7NXBkKBWkaVkAq17E=
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
golang.org/x/net v0.0.0-20210324205630-d1beb07c2056 h1:sANdAef76Ioam9aQUUdcAqricwY/WUaMc4+7LY4eGg8=
golang.org/x/net v0.0.0-20210324205630-d1beb07c2056/go.mod h1:uSPa2vr4CLtc/ILN5odXGNXS6mhrKVzTaCXzk9m6W3k=
golang.org/x/net v0.0.0-20210326220855-61e056675ecf/go.mod h1:uSPa2vr4CLtc/ILN5odXGNXS6mhrKVzTaCXzk9m6W3k=
golang.org/x/net v0.0.0-20210331060903-cb1fcc7394e5/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
golang.org/x/net v0.0.0-20210331212208-0fccb6fa2b5c h1:KHUzaHIpjWVlVVNh65G3hhuj3KB1HnjY6Cq5cTvRQT8=
golang.org/x/net v0.0.0-20210331212208-0fccb6fa2b5c/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
golang.org/x/net v0.0.0-20210510120150-4163338589ed h1:p9UgmWI9wKpfYmgaV/IZKGdXc5qEK45tDwwwDyjS26I=
golang.org/x/net v0.0.0-20210510120150-4163338589ed/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -79,15 +59,12 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005 h1:pDMpM2zh2MT0kHy037cKlSby2nEhD50SYqwQk76Nm40=
golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210324051608-47abb6519492/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44 h1:Bli41pIlzTzf3KEY06n+xnzK/BESIg2ze4Pgfh/aI8c=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da h1:b3NXsE2LusjYGGjL5bxEVZZORm/YEFFrWFjR8eFrw/c=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
@@ -107,12 +84,8 @@ google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZi
google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
google.golang.org/grpc v1.36.0 h1:o1bcQ6imQMIOpdrO3SWf2z5RV72WbDwdXuK0MDlc8As=
google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
google.golang.org/grpc v1.36.1 h1:cmUfbeGKnz9+2DD/UYsMQXeqbHZqZDs4eQwW0sFOpBY=
google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
google.golang.org/grpc v1.37.0 h1:uSZWeQJX5j11bIQ4AJoj+McDBo29cY1MCoC1wO3ts+c=
google.golang.org/grpc v1.37.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
google.golang.org/grpc v1.37.1 h1:ARnQJNWxGyYJpdf/JXscNlQr/uv607ZPU9Z7ogHi+iI=
google.golang.org/grpc v1.37.1/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=

53
grpc.go
View File

@@ -9,7 +9,6 @@ import (
"reflect"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/unistack-org/micro/v3/broker"
@@ -23,11 +22,13 @@ import (
gmetadata "google.golang.org/grpc/metadata"
)
const (
defaultContentType = "application/grpc+proto"
)
type grpcClient struct {
opts client.Options
codecs map[string]encoding.Codec
pool *pool
once atomic.Value
init bool
sync.RWMutex
}
@@ -35,17 +36,14 @@ type grpcClient struct {
// secure returns the dial option for whether its a secure or insecure connection
func (g *grpcClient) secure(addr string) grpc.DialOption {
// first we check if theres'a tls config
if g.opts.Context != nil {
if v := g.opts.Context.Value(tlsAuth{}); v != nil {
tls := v.(*tls.Config)
creds := credentials.NewTLS(tls)
if g.opts.TLSConfig != nil {
creds := credentials.NewTLS(g.opts.TLSConfig)
// return tls config if it exists
return grpc.WithTransportCredentials(creds)
}
}
// default config
tlsConfig := &tls.Config{}
tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12}
defaultCreds := grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig))
// check if the address is prepended with https
@@ -86,7 +84,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
md := gmetadata.New(header)
ctx = gmetadata.NewOutgoingContext(ctx, md)
cf, err := g.newGRPCCodec(req.ContentType())
cf, err := g.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
@@ -123,7 +121,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
}
defer func() {
// defer execution of release
g.pool.release(addr, cc, grr)
g.pool.release(cc, grr)
}()
ch := make(chan error, 1)
@@ -131,7 +129,8 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
go func() {
grpcCallOptions := []grpc.CallOption{
grpc.ForceCodec(&wrapMicroCodec{cf}),
grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name())}
grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name()),
}
if opts := g.getGrpcCallOptions(); opts != nil {
grpcCallOptions = append(grpcCallOptions, opts...)
}
@@ -171,7 +170,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
md := gmetadata.New(header)
ctx = gmetadata.NewOutgoingContext(ctx, md)
cf, err := g.newGRPCCodec(req.ContentType())
cf, err := g.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
@@ -187,8 +186,15 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
wc := &wrapMicroCodec{cf}
maxRecvMsgSize := g.maxRecvMsgSizeValue()
maxSendMsgSize := g.maxSendMsgSizeValue()
grpcDialOptions := []grpc.DialOption{
g.secure(addr),
grpc.WithDefaultCallOptions(
grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
grpc.MaxCallSendMsgSize(maxSendMsgSize),
),
}
if opts := g.getGrpcDialOptions(); opts != nil {
@@ -223,7 +229,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
// cancel the context
cancel()
// release the connection
g.pool.release(addr, cc, err)
g.pool.release(cc, err)
// now return the error
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
}
@@ -245,13 +251,13 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
},
conn: cc,
close: func(err error) {
// cancel the context if an error occured
// cancel the context if an error occurred
if err != nil {
cancel()
}
// defer execution of release
g.pool.release(addr, cc, err)
g.pool.release(cc, err)
},
}
@@ -305,10 +311,14 @@ func (g *grpcClient) maxSendMsgSizeValue() int {
return v.(int)
}
func (g *grpcClient) newGRPCCodec(ct string) (codec.Codec, error) {
func (g *grpcClient) newCodec(ct string) (codec.Codec, error) {
g.RLock()
defer g.RUnlock()
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
if c, ok := g.opts.Codecs[ct]; ok {
return c, nil
}
@@ -648,7 +658,7 @@ func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...clie
body = d.Data
} else {
// use codec for payload
cf, err := g.newGRPCCodec(p.ContentType())
cf, err := g.newCodec(p.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
@@ -670,7 +680,10 @@ func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...clie
return g.opts.Broker.Publish(metadata.NewOutgoingContext(ctx, md), topic, &broker.Message{
Header: md,
Body: body,
}, broker.PublishContext(options.Context))
},
broker.PublishContext(options.Context),
broker.PublishBodyOnly(options.BodyOnly),
)
}
func (g *grpcClient) String() string {
@@ -724,7 +737,7 @@ func (g *grpcClient) getGrpcCallOptions() []grpc.CallOption {
func NewClient(opts ...client.Option) client.Client {
options := client.NewOptions(opts...)
// default content type for grpc
options.ContentType = "application/grpc+proto"
options.ContentType = defaultContentType
rc := &grpcClient{
opts: options,

View File

@@ -10,16 +10,12 @@ import (
)
type pool struct {
conns map[string]*streamsPool
size int
ttl int64
// max streams on a *poolConn
maxStreams int
// max idle conns
maxIdle int
sync.Mutex
conns map[string]*streamsPool
}
type streamsPool struct {
@@ -34,20 +30,15 @@ type streamsPool struct {
}
type poolConn struct {
// grpc conn
*grpc.ClientConn
err error
addr string
// pool and streams pool
*grpc.ClientConn
next *poolConn
pool *pool
sp *streamsPool
pre *poolConn
addr string
streams int
created int64
// list
pre *poolConn
next *poolConn
in bool
}
@@ -140,7 +131,7 @@ func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption
if err != nil {
return nil, err
}
conn = &poolConn{cc, nil, addr, p, sp, 1, time.Now().Unix(), nil, nil, 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()
@@ -152,7 +143,7 @@ func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption
return conn, nil
}
func (p *pool) release(addr string, conn *poolConn, err error) {
func (p *pool) release(conn *poolConn, err error) {
p.Lock()
p, sp, created := conn.pool, conn.sp, conn.created
// try to add conn
@@ -188,7 +179,7 @@ func (p *pool) release(addr string, conn *poolConn, err error) {
}
func (conn *poolConn) Close() {
conn.pool.release(conn.addr, conn, conn.err)
conn.pool.release(conn, conn.err)
}
func removeConn(conn *poolConn) {

View File

@@ -5,9 +5,9 @@ import (
)
type grpcEvent struct {
payload interface{}
topic string
contentType string
payload interface{}
}
func newGRPCEvent(topic string, payload interface{}, contentType string, opts ...client.MessageOption) client.Message {

View File

@@ -3,7 +3,6 @@ package grpc
import (
"context"
"crypto/tls"
"github.com/unistack-org/micro/v3/client"
"google.golang.org/grpc"
@@ -29,13 +28,6 @@ var (
)
type poolMaxStreams struct{}
type poolMaxIdle struct{}
type codecsKey struct{}
type tlsAuth struct{}
type maxRecvMsgSizeKey struct{}
type maxSendMsgSizeKey struct{}
type grpcDialOptions struct{}
type grpcCallOptions struct{}
// maximum streams on a connectioin
func PoolMaxStreams(n int) client.Option {
@@ -47,6 +39,8 @@ func PoolMaxStreams(n int) client.Option {
}
}
type poolMaxIdle struct{}
// maximum idle conns of a pool
func PoolMaxIdle(d int) client.Option {
return func(o *client.Options) {
@@ -57,6 +51,8 @@ func PoolMaxIdle(d int) client.Option {
}
}
type codecsKey struct{}
// gRPC Codec to be used to encode/decode requests for a given content type
func Codec(contentType string, c encoding.Codec) client.Option {
return func(o *client.Options) {
@@ -72,15 +68,7 @@ func Codec(contentType string, c encoding.Codec) client.Option {
}
}
// AuthTLS should be used to setup a secure authentication using TLS
func AuthTLS(t *tls.Config) client.Option {
return func(o *client.Options) {
if o.Context == nil {
o.Context = context.Background()
}
o.Context = context.WithValue(o.Context, tlsAuth{}, t)
}
}
type maxRecvMsgSizeKey struct{}
//
// MaxRecvMsgSize set the maximum size of message that client can receive.
@@ -94,6 +82,8 @@ func MaxRecvMsgSize(s int) client.Option {
}
}
type maxSendMsgSizeKey struct{}
//
// MaxSendMsgSize set the maximum size of message that client can send.
//
@@ -106,6 +96,8 @@ func MaxSendMsgSize(s int) client.Option {
}
}
type grpcDialOptions struct{}
//
// DialOptions to be used to configure gRPC dial options
//
@@ -118,6 +110,8 @@ func DialOptions(opts ...grpc.DialOption) client.CallOption {
}
}
type grpcCallOptions struct{}
//
// CallOptions to be used to configure gRPC call options
//

View File

@@ -9,12 +9,12 @@ import (
)
type grpcRequest struct {
request interface{}
codec codec.Codec
service string
method string
contentType string
request interface{}
opts client.RequestOptions
codec codec.Codec
}
// service Struct.Method /service.Struct/Method

View File

@@ -11,17 +11,15 @@ import (
// Implements the streamer interface
type grpcStream struct {
// embed so we can access if need be
grpc.ClientStream
sync.RWMutex
closed bool
context context.Context
err error
conn *poolConn
request client.Request
response client.Response
context context.Context
close func(err error)
conn *poolConn
sync.RWMutex
closed bool
}
func (g *grpcStream) Context() context.Context {