Compare commits
17 Commits
Author | SHA1 | Date | |
---|---|---|---|
1206fecf93 | |||
e091caef5d | |||
d8508301cf | |||
|
03dd750f7c | ||
|
5914e610a1 | ||
|
48841609ba | ||
|
dfc0ac0ea8 | ||
d0bc3d0d3b | |||
683eb62973 | |||
|
b0cad975e4 | ||
|
488d3995e7 | ||
8eeab44898 | |||
|
668e034041 | ||
|
366afe2374 | ||
|
0812a98bf9 | ||
|
eab671950b | ||
|
8409d351ea |
13
.github/stale.sh
vendored
13
.github/stale.sh
vendored
@@ -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
44
.golangci.yml
Normal 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
36
codec.go
@@ -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
9
error.go
@@ -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
|
||||
}
|
||||
|
||||
|
8
go.mod
8
go.mod
@@ -1,14 +1,12 @@
|
||||
module github.com/unistack-org/micro-server-grpc/v3
|
||||
|
||||
go 1.15
|
||||
go 1.16
|
||||
|
||||
require (
|
||||
github.com/golang/protobuf v1.5.2
|
||||
github.com/unistack-org/micro/v3 v3.3.11
|
||||
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
|
||||
github.com/unistack-org/micro/v3 v3.3.17
|
||||
golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
|
||||
google.golang.org/grpc v1.37.0
|
||||
google.golang.org/protobuf v1.26.0
|
||||
)
|
||||
|
||||
//replace github.com/unistack-org/micro/v3 => ../../micro
|
||||
|
18
go.sum
18
go.sum
@@ -42,8 +42,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.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.17 h1:WcyS7InP0DlS/JpRQGLh5sG6VstkdHJbgpMp+gmHmwg=
|
||||
github.com/unistack-org/micro/v3 v3.3.17/go.mod h1:022EOEZZ789hZY3yB5ZSMXU6jLiadBgcNB/cpediV3c=
|
||||
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=
|
||||
@@ -53,10 +53,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-20210331212208-0fccb6fa2b5c h1:KHUzaHIpjWVlVVNh65G3hhuj3KB1HnjY6Cq5cTvRQT8=
|
||||
golang.org/x/net v0.0.0-20210331212208-0fccb6fa2b5c/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0=
|
||||
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||
golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6 h1:0PC75Fz/kyMGhL0e1QnypqK2kQMqKt9csD1GnMJR+Zk=
|
||||
golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
|
||||
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=
|
||||
@@ -64,12 +62,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-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=
|
||||
|
89
grpc.go
89
grpc.go
@@ -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 {
|
||||
@@ -188,10 +183,8 @@ func (g *grpcServer) getMaxMsgSize() int {
|
||||
}
|
||||
|
||||
func (g *grpcServer) getCredentials() credentials.TransportCredentials {
|
||||
if g.opts.Context != nil {
|
||||
if v, ok := g.opts.Context.Value(tlsAuth{}).(*tls.Config); ok && v != nil {
|
||||
return credentials.NewTLS(v)
|
||||
}
|
||||
if g.opts.TLSConfig != nil {
|
||||
return credentials.NewTLS(g.opts.TLSConfig)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -259,7 +252,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
|
||||
// timeout for server deadline
|
||||
to, ok := md.Get("timeout")
|
||||
if ok {
|
||||
md.Del("x-content-type")
|
||||
md.Del("timeout")
|
||||
}
|
||||
|
||||
// get content type
|
||||
@@ -294,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{}
|
||||
@@ -313,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()
|
||||
@@ -325,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.
|
||||
@@ -433,9 +428,10 @@ func (g *grpcServer) processRequest(stream grpc.ServerStream, service *service,
|
||||
}
|
||||
|
||||
return status.New(statusCode, statusDesc).Err()
|
||||
}
|
||||
// }
|
||||
}
|
||||
|
||||
/*
|
||||
type reflectStream struct {
|
||||
stream server.Stream
|
||||
}
|
||||
@@ -477,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{
|
||||
@@ -549,6 +546,10 @@ func (g *grpcServer) 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
|
||||
}
|
||||
@@ -572,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 {
|
||||
@@ -633,23 +634,19 @@ func (g *grpcServer) Register() error {
|
||||
|
||||
g.RLock()
|
||||
// Maps are ordered randomly, sort the keys for consistency
|
||||
var handlerList []string
|
||||
for n, e := range g.handlers {
|
||||
handlerList := make([]string, 0, len(g.handlers))
|
||||
for n := range g.handlers {
|
||||
// Only advertise non internal handlers
|
||||
if !e.Options().Internal {
|
||||
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
|
||||
if !e.Options().Internal {
|
||||
subscriberList = append(subscriberList, e)
|
||||
}
|
||||
}
|
||||
sort.Slice(subscriberList, func(i, j int) bool {
|
||||
return subscriberList[i].topic > subscriberList[j].topic
|
||||
})
|
||||
@@ -664,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()
|
||||
@@ -673,7 +670,7 @@ func (g *grpcServer) Register() error {
|
||||
|
||||
if !registered {
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].Id)
|
||||
config.Logger.Infof(config.Context, "Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -733,7 +730,7 @@ func (g *grpcServer) Deregister() error {
|
||||
}
|
||||
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Deregistering node: %s", service.Nodes[0].Id)
|
||||
config.Logger.Infof(config.Context, "Deregistering node: %s", service.Nodes[0].ID)
|
||||
}
|
||||
|
||||
if err := server.DefaultDeregisterFunc(service, config); err != nil {
|
||||
@@ -838,9 +835,10 @@ 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)
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.ID, err)
|
||||
}
|
||||
} else {
|
||||
// announce self to the world
|
||||
@@ -886,25 +884,26 @@ 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)
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s, deregister it", config.Name, config.ID, rerr)
|
||||
}
|
||||
// deregister self in case of error
|
||||
if err := g.Deregister(); err != nil {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s deregister error: %s", config.Name, config.Id, err)
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s deregister error: %s", config.Name, config.ID, err)
|
||||
}
|
||||
}
|
||||
} else if rerr != nil && !registered {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.Id, rerr)
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.ID, rerr)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := g.Register(); err != nil {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register error: %s", config.Name, config.Id, err)
|
||||
config.Logger.Errorf(config.Context, "Server %s-%s register error: %s", config.Name, config.ID, err)
|
||||
}
|
||||
}
|
||||
// wait for exit
|
||||
@@ -916,7 +915,7 @@ func (g *grpcServer) Start() error {
|
||||
// deregister self
|
||||
if err := g.Deregister(); err != nil {
|
||||
if config.Logger.V(logger.ErrorLevel) {
|
||||
config.Logger.Errorf(config.Context, "Server deregister error: ", err)
|
||||
config.Logger.Errorf(config.Context, "Server deregister error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -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)
|
||||
|
17
options.go
17
options.go
@@ -2,18 +2,18 @@ package grpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
|
||||
"github.com/unistack-org/micro/v3/server"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/encoding"
|
||||
)
|
||||
|
||||
type codecsKey struct{}
|
||||
type grpcOptions struct{}
|
||||
type maxMsgSizeKey struct{}
|
||||
type tlsAuth 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 {
|
||||
@@ -30,11 +30,6 @@ func Codec(contentType string, c encoding.Codec) server.Option {
|
||||
}
|
||||
}
|
||||
|
||||
// AuthTLS should be used to setup a secure authentication using TLS
|
||||
func AuthTLS(t *tls.Config) server.Option {
|
||||
return setServerOption(tlsAuth{}, t)
|
||||
}
|
||||
|
||||
// Options to be used to configure gRPC options
|
||||
func Options(opts ...grpc.ServerOption) server.Option {
|
||||
return setServerOption(grpcOptions{}, opts)
|
||||
|
@@ -1,3 +1,5 @@
|
||||
// +build ignore
|
||||
|
||||
/*
|
||||
*
|
||||
* Copyright 2016 gRPC authors.
|
||||
|
25
request.go
25
request.go
@@ -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 {
|
||||
|
@@ -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
31
server.go
@@ -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)
|
||||
|
@@ -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
2
util.go
@@ -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], ".")
|
||||
}
|
||||
*/
|
||||
|
Reference in New Issue
Block a user