添加queue的错误次数限制

Author: wenjianzhang

.gitignore

@@ -18,4 +18,4 @@
 /.idea
 
 /plugins/logger/zap/testdata/*
-go.sum
+go.sum

sdk/antd_api/api.go

@@ -131,3 +131,7 @@ func (e *Api) AddError(err error) {
 	}
 }
 
+func (e Api) Translate(form, to interface{}) {
+	pkg.Translate(form, to)
+}
+

sdk/pkg/jwtauth/jwtauth.go

@@ -65,6 +65,9 @@ type GinJWTMiddleware struct {
 	// User can define own LoginResponse func.
 	LoginResponse func(*gin.Context, int, string, time.Time)
 
+	// User can define own AntdLoginResponse func.
+	AntdLoginResponse func(*gin.Context, int, string, time.Time)
+
 	// User can define own RefreshResponse func.
 	RefreshResponse func(*gin.Context, int, string, time.Time)
 
@@ -213,19 +216,19 @@ var (
 	RKey = "r"
 
 	// RoleIdKey 角色id  Old
-	RoleIdKey   = "roleid"
+	RoleIdKey = "roleid"
 
 	// RoleKey 角色名称  Old
-	RoleKey     = "rolekey"
+	RoleKey = "rolekey"
 
 	// RoleNameKey 角色名称  Old
 	RoleNameKey = "rolename"
 
 	// RoleIdKey 部门id
-	DeptId   = "deptId"
+	DeptId = "deptId"
 
 	// RoleKey 部门名称
-	DeptName     = "deptName"
+	DeptName = "deptName"
 )
 
 // New for check error with GinJWTMiddleware
@@ -328,6 +331,18 @@ func (mw *GinJWTMiddleware) MiddlewareInit() error {
 		}
 	}
 
+	if mw.AntdLoginResponse == nil {
+		mw.AntdLoginResponse = func(c *gin.Context, code int, token string, expire time.Time) {
+			c.JSON(http.StatusOK, gin.H{
+				"code":             http.StatusOK,
+				"success":          true,
+				"token":            token,
+				"currentAuthority": token,
+				"expire":           expire.Format(time.RFC3339),
+			})
+		}
+	}
+
 	if mw.RefreshResponse == nil {
 		mw.RefreshResponse = func(c *gin.Context, code int, token string, expire time.Time) {
 			c.JSON(http.StatusOK, gin.H{
@@ -446,24 +461,19 @@ func (mw *GinJWTMiddleware) LoginHandler(c *gin.Context) {
 		mw.unauthorized(c, http.StatusInternalServerError, mw.HTTPStatusMessageFunc(ErrMissingAuthenticatorFunc, c))
 		return
 	}
-
 	data, err := mw.Authenticator(c)
-
 	if err != nil {
 		mw.unauthorized(c, 400, mw.HTTPStatusMessageFunc(err, c))
 		return
 	}
-
 	// Create the token
 	token := jwt.New(jwt.GetSigningMethod(mw.SigningAlgorithm))
 	claims := token.Claims.(jwt.MapClaims)
-
 	if mw.PayloadFunc != nil {
 		for key, value := range mw.PayloadFunc(data) {
 			claims[key] = value
 		}
 	}
-
 	expire := mw.TimeFunc().Add(mw.Timeout)
 	claims["exp"] = expire.Unix()
 	claims["orig_iat"] = mw.TimeFunc().Unix()
@@ -488,7 +498,7 @@ func (mw *GinJWTMiddleware) LoginHandler(c *gin.Context) {
 		)
 	}
 
-	mw.LoginResponse(c, http.StatusOK, tokenString, expire)
+	mw.AntdLoginResponse(c, http.StatusOK, tokenString, expire)
 }
 
 func (mw *GinJWTMiddleware) signedString(token *jwt.Token) (string, error) {
@@ -747,4 +757,4 @@ func GetToken(c *gin.Context) string {
 	}
 
 	return token.(string)
-}
+}

storage/queue/memory.go

@@ -43,11 +43,16 @@ func (m *Memory) Append(message storage.Messager) error {
 	memoryMessage.SetID(message.GetID())
 	memoryMessage.SetStream(message.GetStream())
 	memoryMessage.SetValues(message.GetValues())
+
 	v, ok := m.queue.Load(message.GetStream())
-	if !ok {
+
+	// TODO: 错误超出5次将放弃
+	if !ok && memoryMessage.GetErrorCount() < 5 {
 		v = m.makeQueue()
 		m.queue.Store(message.GetStream(), v)
+		memoryMessage.SetErrorCount()
 	}
+
 	var q queue
 	switch v.(type) {
 	case queue:

storage/queue/message.go

@@ -8,6 +8,7 @@ import (
 
 type Message struct {
 	redisqueue.Message
+	ErrorCount int
 }
 
 func (m *Message) GetID() string {
@@ -49,3 +50,11 @@ func (m *Message) SetPrefix(prefix string) {
 	}
 	m.Values[storage.PrefixKey] = prefix
 }
+
+func (m *Message) SetErrorCount() {
+	m.ErrorCount = m.ErrorCount + 1
+}
+
+func (m *Message) GetErrorCount() int {
+	return m.ErrorCount
+}

storage/type.go

@@ -39,6 +39,8 @@ type Messager interface {
 	GetValues() map[string]interface{}
 	GetPrefix() string
 	SetPrefix(string)
+	SetErrorCount()
+	GetErrorCount() int
 }
 
 type ConsumerFunc func(Messager) error