fix:

	update middleware
	update errors

Author: kooksee

clients/grpcc/middleware.go

@@ -4,26 +4,27 @@ import (
 	"context"
 	"time"
 
+	"github.com/pubgo/xerror"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
 
+	"github.com/pubgo/lava/pkg/encoding"
 	"github.com/pubgo/lava/types"
 )
 
 func unaryInterceptor(middlewares []types.Middleware) grpc.UnaryClientInterceptor {
-	var wrapperUnary = func(ctx context.Context, req types.Request, rsp func(response types.Response) error) error {
+	var unaryWrapper = func(ctx context.Context, req types.Request, rsp func(response types.Response) error) error {
 		var reqCtx = req.(*request)
 		ctx = metadata.NewOutgoingContext(ctx, reqCtx.Header())
 		if err := reqCtx.invoker(ctx, reqCtx.method, reqCtx.req, reqCtx.reply, reqCtx.cc); err != nil {
 			return err
 		}
-
 		return rsp(&response{req: reqCtx, resp: reqCtx.reply})
 	}
 
 	for i := len(middlewares) - 1; i >= 0; i-- {
-		wrapperUnary = middlewares[i](wrapperUnary)
+		unaryWrapper = middlewares[i](unaryWrapper)
 	}
 
 	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
@@ -58,14 +59,17 @@ func unaryInterceptor(middlewares []types.Middleware) grpc.UnaryClientIntercepto
 			if dur, err := time.ParseDuration(to[0]); err == nil {
 				var cancel context.CancelFunc
 				ctx, cancel = context.WithTimeout(ctx, dur)
-				_ = cancel
+				defer cancel()
 			}
 		}
 
-		return wrapperUnary(
-			ctx,
+		var cdc = encoding.GetCdc(ct)
+		xerror.Assert(cdc == nil, "contentType(%s) codec not found", ct)
+
+		return unaryWrapper(ctx,
 			&request{
 				ct:      ct,
+				cdc:     cdc,
 				header:  md,
 				service: serviceFromMethod(method),
 				opts:    opts,
@@ -128,14 +132,17 @@ func streamInterceptor(middlewares []types.Middleware) grpc.StreamClientIntercep
 			if dur, err := time.ParseDuration(to[0]); err == nil {
 				var cancel context.CancelFunc
 				ctx, cancel = context.WithTimeout(ctx, dur)
-				_ = cancel
+				defer cancel()
 			}
 		}
 
-		return nil, wrapperStream(
-			ctx,
+		var cdc = encoding.GetCdc(ct)
+		xerror.Assert(cdc == nil, "contentType(%s) codec not found", ct)
+
+		return nil, wrapperStream(ctx,
 			&request{
 				ct:       ct,
+				cdc:      cdc,
 				service:  serviceFromMethod(method),
 				header:   md,
 				opts:     opts,
@@ -144,10 +151,7 @@ func streamInterceptor(middlewares []types.Middleware) grpc.StreamClientIntercep
 				method:   method,
 				streamer: streamer,
 			},
-			func(rsp types.Response) error {
-				resp = rsp.(*response).stream
-				return nil
-			},
+			func(rsp types.Response) error { resp = rsp.(*response).stream; return nil },
 		)
 	}
 }

clients/grpcc/request.go

@@ -11,6 +11,7 @@ var _ types.Request = (*request)(nil)
 
 type request struct {
 	ct         string
+	cdc        encoding.Codec
 	opts       []grpc.CallOption
 	method     string
 	service    string
@@ -22,46 +23,14 @@ type request struct {
 	header     types.Header
 }
 
-func (r *request) Kind() string {
-	return Name
-}
-
-func (r *request) Codec() encoding.Codec {
-	return encoding.Get(encoding.Mapping[r.ct])
-}
-
-func (r *request) Client() bool {
-	return true
-}
-
-func (r *request) Service() string {
-	return r.service
-}
-
-func (r *request) Method() string {
-	return r.method
-}
-
-func (r *request) Endpoint() string {
-	return r.method
-}
-
-func (r *request) ContentType() string {
-	return r.ct
-}
-
-func (r *request) Header() types.Header {
-	return r.header
-}
-
-func (r *request) Payload() interface{} {
-	return r.req
-}
-
-func (r *request) Body() ([]byte, error) {
-	return nil, nil
-}
-
-func (r *request) Stream() bool {
-	return false
-}
+func (r *request) Kind() string          { return Name }
+func (r *request) Codec() encoding.Codec { return r.cdc }
+func (r *request) Client() bool          { return true }
+func (r *request) Service() string       { return r.service }
+func (r *request) Method() string        { return r.method }
+func (r *request) Endpoint() string      { return r.method }
+func (r *request) ContentType() string   { return r.ct }
+func (r *request) Header() types.Header  { return r.header }
+func (r *request) Payload() interface{}  { return r.req }
+func (r *request) Read() ([]byte, error) { return nil, nil }
+func (r *request) Stream() bool          { return r.desc != nil }

clients/grpcc/response.go

@@ -1,8 +1,6 @@
 package grpcc
 
 import (
-	"encoding/json"
-
 	"github.com/pubgo/lava/types"
 	"google.golang.org/grpc"
 )
@@ -15,18 +13,6 @@ type response struct {
 	resp   interface{}
 }
 
-func (r *response) Stream() bool {
-	return r.stream != nil
-}
-
-func (r *response) Header() types.Header {
-	return r.req.header
-}
-
-func (r *response) Body() ([]byte, error) {
-	return json.Marshal(r.resp)
-}
-
-func (r *response) Payload() interface{} {
-	return r.resp
-}
+func (r *response) Stream() bool         { return r.stream != nil }
+func (r *response) Header() types.Header { return r.req.header }
+func (r *response) Payload() interface{} { return r.resp }

clients/orm/config.go

@@ -11,6 +11,7 @@ import (
 	"github.com/pubgo/lava/internal/logz"
 	"github.com/pubgo/lava/pkg/merge"
 	"github.com/pubgo/lava/plugins/tracing"
+	"github.com/pubgo/lava/runenv"
 )
 
 var logs = logz.New(Name)
@@ -36,20 +37,26 @@ type Cfg struct {
 
 func (t Cfg) Build(dialect gorm.Dialector) *gorm.DB {
 	var log = merge.Struct(&gorm.Config{}, t).(*gorm.Config)
+
+	var level = logger.Info
+	if runenv.IsProd() || runenv.IsRelease() {
+		level = logger.Error
+	}
+
 	log.Logger = logger.New(
 		logPrintf(logs.Infof),
 		logger.Config{
-			SlowThreshold:             time.Second,   // Slow SQL threshold
-			LogLevel:                  logger.Silent, // Log level
-			IgnoreRecordNotFoundError: true,          // Ignore ErrRecordNotFound error for logger
-			Colorful:                  false,         // Disable color
+			SlowThreshold:             200 * time.Millisecond,
+			LogLevel:                  level,
+			IgnoreRecordNotFoundError: false,
+			Colorful:                  true,
 		},
 	)
 
 	db, err := gorm.Open(dialect, log)
 	xerror.Panic(err)
 
-	// 添加链路
+	// 添加链路追踪
 	xerror.Panic(db.Use(opentracing.New(
 		opentracing.WithErrorTagHook(tracing.SetIfErr),
 	)))

clients/restc/client.go

@@ -4,15 +4,16 @@ import (
 	"bytes"
 	"context"
 	"io"
+	"io/ioutil"
 	"net/http"
 	"net/url"
 	"time"
 
 	"github.com/pubgo/x/strutil"
 	"github.com/pubgo/xerror"
 
+	"github.com/pubgo/lava/pkg/encoding"
 	"github.com/pubgo/lava/pkg/httpx"
-	"github.com/pubgo/lava/pkg/retry"
 	"github.com/pubgo/lava/types"
 )
 
@@ -32,10 +33,25 @@ type clientImpl struct {
 }
 
 func (c *clientImpl) Do(ctx context.Context, req *Request) (resp *Response, err error) {
-	return resp, c.do(ctx, &request{req: req}, func(res types.Response) error {
-		resp = res.(*response).resp
-		return nil
-	})
+	var ct = filterFlags(req.Header.Get(httpx.HeaderContentType))
+	var cdc = encoding.GetCdc(ct)
+	xerror.Assert(cdc == nil, "contentType(%s) codec not found", ct)
+
+	var data []byte
+	if req.Body != nil {
+		data, err = ioutil.ReadAll(req.Body)
+		xerror.Panic(err)
+		req.Body = ioutil.NopCloser(bytes.NewReader(data))
+	}
+
+	return resp, c.do(ctx,
+		&request{
+			ct:  ct,
+			cdc: cdc,
+			req: req,
+		},
+		func(res types.Response) error { resp = res.(*response).resp; return nil },
+	)
 }
 
 func (c *clientImpl) Get(ctx context.Context, url string, requests ...func(req *Request)) (*Response, error) {
@@ -92,15 +108,11 @@ func doRequest(ctx context.Context, c *clientImpl, mth string, url string, reque
 	return resp, nil
 }
 
-func doFunc(c *clientImpl) types.MiddleNext {
-	var r = retry.New(retry.WithMaxRetries(c.cfg.RetryCount, c.cfg.backoff))
-	return func(ctx context.Context, req types.Request, callback func(rsp types.Response) error) error {
-		return r.Do(func(i int) error {
-			resp, err := c.client.Do(req.(*request).req)
-			if err != nil {
-				return err
-			}
-			return callback(&response{resp: resp})
-		})
+func filterFlags(content string) string {
+	for i, char := range content {
+		if char == ' ' || char == ';' {
+			return content[:i]
+		}
 	}
+	return content
 }

clients/restc/config.go

@@ -69,7 +69,7 @@ func (t Cfg) Build(opts ...func(cfg *Cfg)) (_ Client, err error) {
 		if mid == nil {
 			continue
 		}
-		middlewares = append(middlewares, plugin.Get(name).Middleware())
+		middlewares = append(middlewares, mid)
 	}
 
 	// 加载插件

clients/restc/middleware.go

@@ -0,0 +1,21 @@
+package restc
+
+import (
+	"context"
+	
+	"github.com/pubgo/lava/pkg/retry"
+	"github.com/pubgo/lava/types"
+)
+
+func doFunc(c *clientImpl) types.MiddleNext {
+	var r = retry.New(retry.WithMaxRetries(c.cfg.RetryCount, c.cfg.backoff))
+	return func(ctx context.Context, req types.Request, callback func(rsp types.Response) error) error {
+		return r.Do(func(i int) error {
+			resp, err := c.client.Do(req.(*request).req)
+			if err != nil {
+				return err
+			}
+			return callback(&response{resp: resp})
+		})
+	}
+}

clients/restc/request.go

@@ -8,49 +8,20 @@ import (
 var _ types.Request = (*request)(nil)
 
 type request struct {
-	req    *Request
-}
-
-func (r *request) Kind() string {
-	return Name
-}
-
-func (r *request) Codec() encoding.Codec {
-	return encoding.Get(encoding.Mapping[r.ContentType()])
-}
-
-func (r *request) Client() bool {
-	return true
-}
-
-func (r *request) Service() string {
-	return ""
-}
-
-func (r *request) Method() string {
-	return r.req.Method
-}
-
-func (r *request) Endpoint() string {
-	return r.req.RequestURI
-}
-
-func (r *request) ContentType() string {
-	return r.ContentType()
-}
-
-func (r *request) Header() types.Header {
-	return types.Header(r.req.Header)
-}
-
-func (r *request) Payload() interface{} {
-	return nil
-}
-
-func (r *request) Body() ([]byte, error) {
-	return nil, nil
-}
-
-func (r *request) Stream() bool {
-	return false
-}
+	service string
+	req     *Request
+	ct      string
+	cdc     encoding.Codec
+}
+
+func (r *request) Kind() string          { return Name }
+func (r *request) Codec() encoding.Codec { return r.cdc }
+func (r *request) Client() bool          { return true }
+func (r *request) Service() string       { return r.service }
+func (r *request) Method() string        { return r.req.Method }
+func (r *request) Endpoint() string      { return r.req.RequestURI }
+func (r *request) ContentType() string   { return r.ct }
+func (r *request) Header() types.Header  { return types.Header(r.req.Header) }
+func (r *request) Payload() interface{}  { return nil }
+func (r *request) Read() ([]byte, error) { return nil, nil }
+func (r *request) Stream() bool          { return false }