unistack-org/micro-client-grpc
4
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases 35 Wiki Activity

lint fixes #32

Merged
vtolstov merged 2 commits from lint into master 2021-04-26 19:17:29 +03:00
Conversation 0 Commits 2 Files Changed 9 +93 -73
9 changed files with 93 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

19
codec.go
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
}

22
grpc.go
View File

@ -9,7 +9,6 @@ import (
"reflect"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/unistack-org/micro/v3/broker"
@ -24,11 +23,9 @@ import (
)
type grpcClient struct {
opts client.Options
codecs map[string]encoding.Codec
pool *pool
once atomic.Value
init bool
opts client.Options
pool *pool
init bool
sync.RWMutex
}
@ -42,7 +39,7 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
}
// 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
@ -120,7 +117,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)
@ -128,7 +125,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...)
}
@ -220,7 +218,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))
}
@ -242,13 +240,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)
},
}

33
grpc_pool.go
View File

@ -10,16 +10,12 @@ import (
)
type pool struct {
size int
ttl int64
// max streams on a *poolConn
conns map[string]*streamsPool
size int
ttl int64
maxStreams int
// max idle conns
maxIdle int
maxIdle int
sync.Mutex
conns map[string]*streamsPool
}
type streamsPool struct {
@ -34,21 +30,16 @@ type streamsPool struct {
}
type poolConn struct {
// grpc conn
err error
*grpc.ClientConn
err error
addr string
// pool and streams pool
next *poolConn
pool *pool
sp *streamsPool
pre *poolConn
addr string
streams int
created int64
// list
pre *poolConn
next *poolConn
in bool
in bool
}
func newPool(size int, ttl time.Duration, idle int, ms int) *pool {
@ -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) {

2
message.go
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 {

19
options.go
View File

@ -28,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 {
@ -46,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) {
@ -56,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) {
@ -71,6 +68,8 @@ func Codec(contentType string, c encoding.Codec) client.Option {
}
}
type maxRecvMsgSizeKey struct{}
//
// MaxRecvMsgSize set the maximum size of message that client can receive.
//
@ -83,6 +82,8 @@ func MaxRecvMsgSize(s int) client.Option {
}
}
type maxSendMsgSizeKey struct{}
//
// MaxSendMsgSize set the maximum size of message that client can send.
//
@ -95,6 +96,8 @@ func MaxSendMsgSize(s int) client.Option {
}
}
type grpcDialOptions struct{}
//
// DialOptions to be used to configure gRPC dial options
//
@ -107,6 +110,8 @@ func DialOptions(opts ...grpc.DialOption) client.CallOption {
}
}
type grpcCallOptions struct{}
//
// CallOptions to be used to configure gRPC call options
//

4
request.go
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

10
stream.go
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 {