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

lint

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2021-04-26 18:48:54 +03:00
parent d0bc3d0d3b
commit c2eedbe3b5
12 changed files with 261 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

36
codec.go
View File

@ -4,25 +4,9 @@ import (
"io"
"github.com/unistack-org/micro/v3/codec"
"google.golang.org/grpc"
"google.golang.org/grpc/encoding"
)
type wrapStream struct{ grpc.ServerStream }
func (w *wrapStream) Write(d []byte) (int, error) {
n := len(d)
err := w.ServerStream.SendMsg(&codec.Frame{Data: d})
return n, err
}
func (w *wrapStream) Read(d []byte) (int, error) {
m := &codec.Frame{}
err := w.ServerStream.RecvMsg(m)
d = m.Data
return len(d), err
}
type wrapMicroCodec struct{ codec.Codec }
func (w *wrapMicroCodec) Name() string {
@ -36,43 +20,39 @@ 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
}
return w.Codec.Unmarshal(d, v)
}
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 {
return nil
}
func (g *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
// caller has requested a frame
switch m := v.(type) {
case *codec.Frame:
func (w *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
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
}

9
error.go
View File

@ -10,8 +10,7 @@ import (
"google.golang.org/grpc/codes"
)
var (
errMapping = map[int32]codes.Code{
var errMapping = map[int32]codes.Code{
http.StatusOK: codes.OK,
http.StatusBadRequest: codes.InvalidArgument,
http.StatusRequestTimeout: codes.DeadlineExceeded,
@ -23,8 +22,7 @@ var (
http.StatusNotImplemented: codes.Unimplemented,
http.StatusInternalServerError: codes.Internal,
http.StatusServiceUnavailable: codes.Unavailable,
}
)
}
// convertCode converts a standard Go error into its canonical code. Note that
// this is only used to translate the error returned by the server applications.
@ -60,8 +58,7 @@ func microError(err error) codes.Code {
}
var ec int32
switch verr := err.(type) {
case *errors.Error:
if verr, ok := err.(*errors.Error); ok {
ec = verr.Code
}

57
grpc.go
View File

@ -14,6 +14,7 @@ import (
"sync"
"time"
// nolint: staticcheck
oldproto "github.com/golang/protobuf/proto"
"github.com/unistack-org/micro/v3/broker"
"github.com/unistack-org/micro/v3/codec"
@ -29,7 +30,6 @@ import (
"google.golang.org/grpc/encoding"
gmetadata "google.golang.org/grpc/metadata"
"google.golang.org/grpc/peer"
grpcreflect "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
)
@ -38,32 +38,27 @@ const (
defaultContentType = "application/grpc"
)
/*
type grpcServerReflection struct {
srv *grpc.Server
s *serverReflectionServer
}
*/
type grpcServer struct {
rpc *rServer
handlers map[string]server.Handler
srv *grpc.Server
exit chan chan error
wg *sync.WaitGroup
sync.RWMutex
opts server.Options
handlers map[string]server.Handler
subscribers map[*subscriber][]broker.Subscriber
init bool
// marks the serve as started
started bool
// used for first registration
registered bool
reflection bool
// register service instance
rsvc *register.Service
codecs map[string]codec.Codec
subscribers map[*subscriber][]broker.Subscriber
rpc *rServer
opts server.Options
sync.RWMutex
init bool
started bool
registered bool
reflection bool
}
func newGRPCServer(opts ...server.Option) server.Server {
@ -292,6 +287,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
svc := g.rpc.serviceMap[serviceName]
g.rpc.mu.RUnlock()
/*
if svc == nil && g.reflection && methodName == "ServerReflectionInfo" {
rfl := &grpcServerReflection{srv: g.srv, s: &serverReflectionServer{s: g.srv}}
svc = &service{}
@ -311,6 +307,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
g.rpc.mu.Unlock()
}
}
*/
if svc == nil {
return status.New(codes.Unimplemented, fmt.Sprintf("unknown service %s", serviceName)).Err()
@ -323,15 +320,15 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
// process unary
if !mtype.stream {
return g.processRequest(stream, svc, mtype, ct, ctx)
return g.processRequest(ctx, stream, svc, mtype, ct)
}
// process stream
return g.processStream(stream, svc, mtype, ct, ctx)
return g.processStream(ctx, stream, svc, mtype, ct)
}
func (g *grpcServer) processRequest(stream grpc.ServerStream, service *service, mtype *methodType, ct string, ctx context.Context) error {
for {
func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStream, service *service, mtype *methodType, ct string) error {
// for {
var argv, replyv reflect.Value
// Decode the argument value.
@ -431,9 +428,10 @@ func (g *grpcServer) processRequest(stream grpc.ServerStream, service *service,
}
return status.New(statusCode, statusDesc).Err()
}
// }
}
/*
type reflectStream struct {
stream server.Stream
}
@ -475,8 +473,9 @@ func (s *reflectStream) RecvMsg(m interface{}) error {
func (g *grpcServerReflection) ServerReflectionInfo(ctx context.Context, stream server.Stream) error {
return g.s.ServerReflectionInfo(&reflectStream{stream})
}
*/
func (g *grpcServer) processStream(stream grpc.ServerStream, service *service, mtype *methodType, ct string, ctx context.Context) error {
func (g *grpcServer) processStream(ctx context.Context, stream grpc.ServerStream, service *service, mtype *methodType, ct string) error {
opts := g.opts
r := &rpcRequest{
@ -574,7 +573,7 @@ func (g *grpcServer) Init(opts ...server.Option) error {
}
func (g *grpcServer) NewHandler(h interface{}, opts ...server.HandlerOption) server.Handler {
return newRpcHandler(h, opts...)
return newRPCHandler(h, opts...)
}
func (g *grpcServer) Handle(h server.Handler) error {
@ -635,15 +634,15 @@ func (g *grpcServer) Register() error {
g.RLock()
// Maps are ordered randomly, sort the keys for consistency
var handlerList []string
for n, _ := range g.handlers {
handlerList := make([]string, 0, len(g.handlers))
for n := range g.handlers {
// Only advertise non internal handlers
handlerList = append(handlerList, n)
}
sort.Strings(handlerList)
var subscriberList []*subscriber
subscriberList := make([]*subscriber, 0, len(g.subscribers))
for e := range g.subscribers {
// Only advertise non internal subscribers
subscriberList = append(subscriberList, e)
@ -662,7 +661,7 @@ func (g *grpcServer) Register() error {
g.RUnlock()
service.Nodes[0].Metadata["protocol"] = "grpc"
service.Nodes[0].Metadata["transport"] = "grpc"
service.Nodes[0].Metadata["transport"] = service.Nodes[0].Metadata["protocol"]
service.Endpoints = endpoints
g.RLock()
@ -836,6 +835,7 @@ func (g *grpcServer) Start() error {
}
// use RegisterCheck func before register
// nolint: nestif
if err := g.opts.RegisterCheck(config.Context); err != nil {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.Id, err)
@ -884,6 +884,7 @@ func (g *grpcServer) Start() error {
registered := g.registered
g.RUnlock()
rerr := g.opts.RegisterCheck(g.opts.Context)
// nolint: nestif
if rerr != nil && registered {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s, deregister it", config.Name, config.Id, rerr)

8
handler.go
View File

@ -8,13 +8,13 @@ import (
)
type rpcHandler struct {
name string
handler interface{}
endpoints []*register.Endpoint
opts server.HandlerOptions
handler interface{}
name string
endpoints []*register.Endpoint
}
func newRpcHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
func newRPCHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
options := server.NewHandlerOptions(opts...)
typ := reflect.TypeOf(handler)

10
options.go
View File

@ -8,10 +8,12 @@ import (
"google.golang.org/grpc/encoding"
)
type codecsKey struct{}
type grpcOptions struct{}
type maxMsgSizeKey struct{}
type reflectionKey struct{}
type (
codecsKey struct{}
grpcOptions struct{}
maxMsgSizeKey struct{}
reflectionKey struct{}
)
// gRPC Codec to be used to encode/decode requests for a given content type
func Codec(contentType string, c encoding.Codec) server.Option {

2
reflection.go
View File

@ -1,3 +1,5 @@
// +build ignore
/*
*
* Copyright 2016 gRPC authors.

25
request.go
View File

@ -5,29 +5,34 @@ import (
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/metadata"
"github.com/unistack-org/micro/v3/server"
)
var (
_ server.Request = &rpcRequest{}
_ server.Message = &rpcMessage{}
)
type rpcRequest struct {
rw io.ReadWriter
service string
method string
endpoint string
target string
contentType string
payload interface{}
codec codec.Codec
header metadata.Metadata
method string
endpoint string
contentType string
service string
body []byte
stream bool
payload interface{}
}
type rpcMessage struct {
payload interface{}
codec codec.Codec
header metadata.Metadata
topic string
contentType string
payload interface{}
header metadata.Metadata
body []byte
codec codec.Codec
}
func (r *rpcRequest) ContentType() string {
@ -43,7 +48,7 @@ func (r *rpcRequest) Method() string {
}
func (r *rpcRequest) Endpoint() string {
return r.method
return r.endpoint
}
func (r *rpcRequest) Codec() codec.Codec {

7
response.go
View File

@ -5,16 +5,15 @@ import (
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/metadata"
"github.com/unistack-org/micro/v3/server"
)
var _ server.Response = &rpcResponse{}
type rpcResponse struct {
rw io.ReadWriter
header metadata.Metadata
codec codec.Codec
endpoint string
service string
method string
target string
}
func (r *rpcResponse) Codec() codec.Codec {

31
server.go
View File

@ -17,34 +17,32 @@ import (
"github.com/unistack-org/micro/v3/server"
)
var (
// Precompute the reflect type for error. Can't use error directly
// because Typeof takes an empty interface value. This is annoying.
typeOfError = reflect.TypeOf((*error)(nil)).Elem()
)
// Precompute the reflect type for error. Can't use error directly
// because Typeof takes an empty interface value. This is annoying.
var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
type methodType struct {
method reflect.Method
ArgType reflect.Type
ReplyType reflect.Type
ContextType reflect.Type
method reflect.Method
stream bool
}
type reflectionType func(context.Context, server.Stream) error
// type reflectionType func(context.Context, server.Stream) error
type service struct {
name string // name of service
rcvr reflect.Value // receiver of methods for the service
typ reflect.Type // type of the receiver
method map[string]*methodType // registered methods
typ reflect.Type
method map[string]*methodType
rcvr reflect.Value
name string
}
// server represents an RPC Server.
type rServer struct {
mu sync.RWMutex // protects the serviceMap
serviceMap map[string]*service
reflection bool
mu sync.RWMutex
// reflection bool
}
// Is this an exported - upper case - name?
@ -91,15 +89,14 @@ func prepareEndpoint(method reflect.Method) (*methodType, error) {
return nil, fmt.Errorf("method %v of %v has wrong number of ins: %v", mname, mtype, mtype.NumIn())
}
if stream {
switch stream {
case true:
// check stream type
streamType := reflect.TypeOf((*server.Stream)(nil)).Elem()
if !argType.Implements(streamType) {
return nil, fmt.Errorf("%v argument does not implement Streamer interface: %v", mname, argType)
}
} else {
// if not stream check the replyType
default:
// First arg need not be a pointer.
if !isExportedOrBuiltinType(argType) {
return nil, fmt.Errorf("%v argument type not exported: %v", mname, argType)

10
subscriber.go
View File

@ -15,14 +15,10 @@ import (
"github.com/unistack-org/micro/v3/server"
)
const (
subSig = "func(context.Context, interface{}) error"
)
type handler struct {
method reflect.Value
reqType reflect.Type
ctxType reflect.Type
method reflect.Value
}
type subscriber struct {
@ -195,14 +191,14 @@ func (g *grpcServer) createSubHandler(sb *subscriber, opts server.Options) broke
if g.wg != nil {
defer g.wg.Done()
}
err := fn(ctx, &rpcMessage{
cerr := fn(ctx, &rpcMessage{
topic: sb.topic,
contentType: ct,
payload: req.Interface(),
header: msg.Header,
body: msg.Body,
})
results <- err
results <- cerr
}()
}
var errors []string

2
util.go
View File

@ -39,6 +39,7 @@ func serviceMethod(m string) (string, string, error) {
return parts[0], parts[1], nil
}
/*
// ServiceFromMethod returns the service
// /service.Foo/Bar => service
func serviceFromMethod(m string) string {
@ -55,3 +56,4 @@ func serviceFromMethod(m string) string {
parts = strings.Split(parts[1], ".")
return strings.Join(parts[:len(parts)-1], ".")
}
*/