Skip to content

X
xpkg
提交 ce21a10a authored 作者: mooncake9527's avatar mooncake9527
浏览文件
操作

v1.0.59

上级 eeaf1350
隐藏空白字符变更
内嵌 并排
正在显示 包含 302 行增加 和 283 行删除
app/app.go
浏览文件 @ ce21a10a
......@@ -58,7 +58,7 @@ func (a *App) Run() {
return a.watch(ctx)
})
fmt.Println(buff)
fmt.Println(LOGO1)
if err := eg.Wait(); err != nil {
panic(err)
......@@ -129,3 +129,81 @@ const (
////////////////////////////////////////////////////////////////////
`
)
const (
LOGO = `
¶ ¶
¶ ¶
¶ ¶ ¶ ¶
¶ ¶¶ ¶¶ ¶
¶¶ ¶¶¶ ¶¶¶ ¶¶
¶ ¶¶ ¶¶¶ ¶¶¶ ¶¶ ¶
¶¶ ¶¶ ¶¶¶ ¶¶¶ ¶¶ ¶¶
¶¶ ¶¶ ¶¶¶¶ ¶¶¶¶ ¶¶ ¶¶
¶¶ ¶¶¶ ¶¶¶¶ ¶¶¶¶¶ ¶¶¶ ¶¶¶
¶ ¶¶¶ ¶¶¶¶ ¶¶¶¶ ¶¶¶¶ ¶¶¶¶ ¶¶¶¶ ¶
¶¶ ¶¶¶¶¶ ¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶ ¶¶¶¶¶ ¶¶
¶¶ ¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶ ¶
¶¶ ¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶ ¶¶
¶¶¶ ¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶ ¶¶¶
¶¶¶¶ ¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶ ¶¶¶¶
¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶
¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶
¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶
¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶
¶¶¶¶¶¶¶ .. ¶¶¶¶¶¶¶¶¶ .. ¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶ ¶¶¶ ¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶¶¶¶¶¶
¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶
¶¶¶ ¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶¶ ¶¶¶
¶¶ ¶¶¶¶ ¶¶¶¶¶ ¶¶¶¶ ¶¶
`
)
const (
LOGO1 = `
▐▒▒░▄
▒▒▒▒▒▒▒▒▄
▐▒▒▒▒▒▒▒▒▒▒▒▄ ▄▄▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒░▄ ▄▄▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄ ▄▒▒▒▒▒▒▒▒▒▒▒▒▒▌
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄ ▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ ▄░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░
▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄ ▄▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄▄▄▄▄░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▀▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▀
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▀▒▒▒░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▀
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░▒▒▒▒▀
▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒░▄▄▄ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░▄▄▄ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒ ▀█▀ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░▐██▀ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒░
▒▒▒▒▒▒▒▒▒▒▒▒ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▐▒▒▒▒▒▒▒▒▒▒▒▒▒▄ ▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄ ▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▀▀▀▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄▄ ▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒░░░░░░░░▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░░░░░░▒▒▒▒▒▒▒▒
▐▒░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░░░░░░░░░▐▒▒▒▒▒▒
▒▒░░░░░░░░░░░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▀░░░░▄░░░▀▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░░░░░░░░░▐▒▒▒▒▒▒▒
▐▒▒░░░░░░░▄░▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ ▒▒▒▒▒▒▒▒▒▒ ▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░░░░░░░░▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ ▒▒▒▒▒▒▒▒▒▒▒░▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░░░░▒▒▒▒▒▒▒▒▒▒▒▒
▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ ▒▒▒▒▒▒▒▒▒▒░▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▄░▀▒▒▒▒▀░▄▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
▐▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
`
)
conf/local_parse.go
浏览文件 @ ce21a10a
......@@ -21,7 +21,6 @@ func Parse(configFile string, obj interface{}, reloads ...func()) error {
if err != nil {
return err
}
filePathStr, filename := filepath.Split(confFileAbs)
ext := strings.TrimLeft(path.Ext(filename), ".")
filename = strings.ReplaceAll(filename, "."+ext, "") // excluding suffix names
......
gin/ctxUtils/ctx.go
浏览文件 @ ce21a10a
......@@ -14,47 +14,41 @@ import (
"go.uber.org/zap"
)
type CtxKey string
func (x CtxKey) String() string {
return string(x)
}
const (
GinContextKey CtxKey = "ginContext"
HeaderXTimestampKey CtxKey = "Timestamp"
KeyReqBody CtxKey = "reqBody"
KeyRspBody CtxKey = "rspBody"
KeyRspCode CtxKey = "rspCode"
KeyOApiReqBody CtxKey = "oApiReqBody%s"
KeyOApiRspBody CtxKey = "oApiRspBody%s"
KeyAppName CtxKey = "appName"
KeyUser CtxKey = "user"
KeyApiStartTime CtxKey = "apiStartTime"
KeyClientIP CtxKey = "clientIP"
KeyUID CtxKey = "uid"
KeyUType CtxKey = "uType"
KeyCompanyID CtxKey = "companyID"
KeyShopID CtxKey = "shopID"
KeyUName CtxKey = "uname"
KeyToken CtxKey = "token"
KeyCost CtxKey = "X-Request-Cost"
KeyRspBodyMax CtxKey = "rsp-body-max"
GinContextKey = "ginContext"
HeaderXTimestampKey = "Timestamp"
KeyReqBody = "reqBody"
KeyRspBody = "rspBody"
KeyRspCode = "rspCode"
KeyOApiReqBody = "oApiReqBody%s"
KeyOApiRspBody = "oApiRspBody%s"
KeyAppName = "appName"
KeyUser = "user"
KeyApiStartTime = "apiStartTime"
KeyClientIP = "clientIP"
KeyUID = "uid"
KeyUType = "uType"
KeyCompanyID = "companyID"
KeyShopID = "shopID"
KeyUName = "uname"
KeyToken = "token"
KeyCost = "X-Request-Cost"
KeyRspBodyMax = "rsp-body-max"
)
var (
HeaderXRequestIDKey CtxKey = "X-Request-ID"
ContextTraceIDKey CtxKey = "X-Request-ID"
HeaderXRequestIDKey = "X-Request-ID"
ContextTraceIDKey = "X-Request-ID"
)
var (
ErrorGinContextNotFound = errors.New("gin context not found")
)
func Set(c *gin.Context, key CtxKey, val any) {
c.Set(key.String(), val)
func Set(c *gin.Context, key string, val any) {
c.Set(key, val)
}
func SetRspBodyMax(ctx context.Context, max int) {
......@@ -63,8 +57,8 @@ func SetRspBodyMax(ctx context.Context, max int) {
}
}
func Get[T any](c *gin.Context, key CtxKey) (T, bool) {
if val, ok := c.Get(key.String()); ok {
func Get[T any](c *gin.Context, key string) (T, bool) {
if val, ok := c.Get(key); ok {
if v, ok1 := val.(T); ok1 {
return v, true
}
......@@ -75,7 +69,7 @@ func Get[T any](c *gin.Context, key CtxKey) (T, bool) {
// GetGinCtxTraceID get request id from gin.Context
func GetGinCtxTraceID(c *gin.Context) string {
if v, isExist := c.Get(ContextTraceIDKey.String()); isExist {
if v, isExist := c.Get(ContextTraceIDKey); isExist {
if requestID, ok := v.(string); ok {
return requestID
}
......@@ -85,7 +79,7 @@ func GetGinCtxTraceID(c *gin.Context) string {
// GinTraceIDField get request id field from gin.Context
func GinTraceIDField(c *gin.Context) zap.Field {
return zap.String(ContextTraceIDKey.String(), GetGinCtxTraceID(c))
return zap.String(ContextTraceIDKey, GetGinCtxTraceID(c))
}
// CtxRequestID get request id from context.Context
......@@ -115,23 +109,27 @@ func CtxGetGinCtx(c context.Context) *gin.Context {
// CtxTraceIDField get request id field from context.Context
func CtxTraceIDField(c context.Context) zap.Field {
return zap.String(ContextTraceIDKey.String(), CtxRequestID(c))
return zap.String(ContextTraceIDKey, CtxRequestID(c))
}
func NewCtx() context.Context {
return context.WithValue(context.Background(), ContextTraceIDKey, GenerateTid())
}
func GetGinUserID(c *gin.Context) xsf.ID {
return getGinVal[xsf.ID](c, KeyUID.String())
return getGinVal[xsf.ID](c, KeyUID)
}
func GetGinShopID(c *gin.Context) xsf.ID {
return getGinVal[xsf.ID](c, KeyShopID.String())
return getGinVal[xsf.ID](c, KeyShopID)
}
func GetCtxUserID(c context.Context) xsf.ID {
return getCtxVal[xsf.ID](c, KeyUID.String())
return getCtxVal[xsf.ID](c, KeyUID)
}
func GetCtxShopID(c context.Context) xsf.ID {
return getCtxVal[xsf.ID](c, KeyShopID.String())
return getCtxVal[xsf.ID](c, KeyShopID)
}
func GetCtxString(c context.Context, key string) string {
val := c.Value(key)
......@@ -150,10 +148,10 @@ func GetCtxUserToken(c context.Context) string {
}
func GetGinCtxTid(c *gin.Context) string {
tid := c.GetString(ContextTraceIDKey.String())
tid := c.GetString(ContextTraceIDKey)
if tid == "" {
tid = GenerateTid()
c.Set(ContextTraceIDKey.String(), tid)
c.Set(ContextTraceIDKey, tid)
}
return tid
}
......@@ -189,9 +187,9 @@ func GetClientIP(ctx context.Context) string {
}
func WrapCtx(c *gin.Context) context.Context {
ctx := context.WithValue(c.Request.Context(), ContextTraceIDKey, c.GetString(ContextTraceIDKey.String())) //nolint
ctx := context.WithValue(c.Request.Context(), ContextTraceIDKey, c.GetString(ContextTraceIDKey))
for k, v := range c.Keys {
ctx = context.WithValue(ctx, k, v) //nolint
ctx = context.WithValue(ctx, k, v)
}
ctx = context.WithValue(ctx, GinContextKey, c) //nolint
return ctx
......@@ -212,15 +210,15 @@ func GetGinCtx(ctx context.Context) (*gin.Context, error) {
}
func GetCtxUID(c context.Context) uint {
return getCtxVal[uint](c, KeyUID.String())
return getCtxVal[uint](c, KeyUID)
}
func GetCtxUType(c context.Context) uint {
return getCtxVal[uint](c, KeyUType.String())
return getCtxVal[uint](c, KeyUType)
}
func GetCtxCompanyID(c context.Context) uint {
return getCtxVal[uint](c, KeyCompanyID.String())
return getCtxVal[uint](c, KeyCompanyID)
}
func getGinVal[T any](c *gin.Context, key string) T {
......@@ -246,7 +244,7 @@ func getCtxVal[T any](c context.Context, key string) T {
}
func GetGinReq(c *gin.Context) []byte {
v, ok1 := c.Get(KeyReqBody.String())
v, ok1 := c.Get(KeyReqBody)
if ok1 {
ret, ok2 := v.([]byte)
if ok2 {
......@@ -257,7 +255,7 @@ func GetGinReq(c *gin.Context) []byte {
}
func GetGinRsp(c *gin.Context) []byte {
v, ok1 := c.Get(KeyRspBody.String())
v, ok1 := c.Get(KeyRspBody)
if ok1 {
ret, ok2 := v.([]byte)
if ok2 {
......@@ -275,14 +273,14 @@ func SetVal(ctx context.Context, key string, val interface{}) context.Context {
}
func AddApiCost(c *gin.Context, appName string, start time.Time) {
cost := c.Writer.Header().Get(KeyCost.String())
cost := c.Writer.Header().Get(KeyCost)
currentCost := fmt.Sprintf("%s:%s", appName, formatDuration(time.Since(start)))
if cost == "" {
cost = currentCost
} else {
cost = fmt.Sprintf("%s,%s", currentCost, cost)
}
c.Writer.Header().Set(KeyCost.String(), cost)
c.Writer.Header().Set(KeyCost, cost)
}
func formatDuration(d time.Duration) string {
......@@ -294,12 +292,12 @@ func SetApiCost(ctx context.Context, header map[string][]string) {
if ginCtx, _ := GetGinCtx(ctx); ginCtx != nil && len(header) > 0 {
cost := ""
for k, v := range header {
if k == KeyCost.String() {
if k == KeyCost {
cost = strings.Join(v, ",")
}
}
if cost != "" {
ginCtx.Writer.Header().Set(KeyCost.String(), cost)
ginCtx.Writer.Header().Set(KeyCost, cost)
}
}
}
gin/middleware/auth.go
浏览文件 @ ce21a10a
......@@ -102,8 +102,8 @@ func Auth(opts ...JwtOption) gin.HandlerFunc {
}
} else {
xsfID, _ := xsf.ParseString(claims.UID)
c.Set(ctxUtils.KeyUID.String(), xsfID)
c.Set(ctxUtils.KeyUName.String(), claims.Name)
c.Set(ctxUtils.KeyUID, xsfID)
c.Set(ctxUtils.KeyUName, claims.Name)
}
c.Next()
......
gin/middleware/logging.go
浏览文件 @ ce21a10a
......@@ -205,19 +205,19 @@ func Logging(opts ...Option) gin.HandlerFunc {
entity.SetCopyReq(c, buf)
c.Set(ctxUtils.KeyAppName.String(), o.appName)
c.Set(ctxUtils.KeyAppName, o.appName)
reqID := ""
if o.traceIDFrom == 1 {
if v, isExist := c.Get(ctxUtils.ContextTraceIDKey.String()); isExist {
if v, isExist := c.Get(ctxUtils.ContextTraceIDKey); isExist {
if requestID, ok := v.(string); ok {
reqID = requestID
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey.String(), reqID))
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey, reqID))
}
}
} else if o.traceIDFrom == 2 {
reqID = c.Request.Header.Get(ctxUtils.HeaderXRequestIDKey.String())
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey.String(), reqID))
reqID = c.Request.Header.Get(ctxUtils.HeaderXRequestIDKey)
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey, reqID))
}
o.log.Info("<<<<<<<<<req", fields...)
......@@ -228,12 +228,12 @@ func Logging(opts ...Option) gin.HandlerFunc {
c.Writer = newWriter
ip := ips.GetClientIP(c)
c.Set(ctxUtils.KeyClientIP.String(), ip)
c.Set(ctxUtils.KeyClientIP, ip)
// processing requests
c.Next()
rspBodyMax := c.GetInt(ctxUtils.KeyRspBodyMax.String())
rspBodyMax := c.GetInt(ctxUtils.KeyRspBodyMax)
contentType := c.Writer.Header().Get("Content-Type")
isMedia := false
if strings.Contains(contentType, "image") {
......@@ -259,7 +259,7 @@ func Logging(opts ...Option) gin.HandlerFunc {
}
}
if reqID != "" {
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey.String(), reqID))
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey, reqID))
}
o.log.Info(">>>>>>>>>rsp", fields...)
......@@ -283,13 +283,13 @@ func SimpleLog(opts ...Option) gin.HandlerFunc {
reqID := ""
if o.traceIDFrom == 1 {
if v, isExist := c.Get(ctxUtils.ContextTraceIDKey.String()); isExist {
if v, isExist := c.Get(ctxUtils.ContextTraceIDKey); isExist {
if requestID, ok := v.(string); ok {
reqID = requestID
}
}
} else if o.traceIDFrom == 2 {
reqID = c.Request.Header.Get(ctxUtils.HeaderXRequestIDKey.String())
reqID = c.Request.Header.Get(ctxUtils.HeaderXRequestIDKey)
}
// processing requests
......@@ -304,7 +304,7 @@ func SimpleLog(opts ...Option) gin.HandlerFunc {
zap.Int("size", c.Writer.Size()),
}
if reqID != "" {
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey.String(), reqID))
fields = append(fields, zap.String(ctxUtils.ContextTraceIDKey, reqID))
}
o.log.Info("Gin msg", fields...)
}
......
gin/middleware/requstid.go
浏览文件 @ ce21a10a
......@@ -20,8 +20,8 @@ type requestIDOptions struct {
func defaultRequestIDOptions() *requestIDOptions {
return &requestIDOptions{
contextRequestIDKey: ctxUtil.ContextTraceIDKey.String(),
headerXRequestIDKey: ctxUtil.HeaderXRequestIDKey.String(),
contextRequestIDKey: ctxUtil.ContextTraceIDKey,
headerXRequestIDKey: ctxUtil.HeaderXRequestIDKey,
}
}
......@@ -32,11 +32,11 @@ func (o *requestIDOptions) apply(opts ...RequestIDOption) {
}
func (o *requestIDOptions) setRequestIDKey() {
if o.contextRequestIDKey != ctxUtil.ContextTraceIDKey.String() {
ctxUtil.ContextTraceIDKey = ctxUtil.CtxKey(o.contextRequestIDKey)
if o.contextRequestIDKey != ctxUtil.ContextTraceIDKey {
ctxUtil.ContextTraceIDKey = o.contextRequestIDKey
}
if o.headerXRequestIDKey != ctxUtil.HeaderXRequestIDKey.String() {
ctxUtil.HeaderXRequestIDKey = ctxUtil.CtxKey(o.headerXRequestIDKey)
if o.headerXRequestIDKey != ctxUtil.HeaderXRequestIDKey {
ctxUtil.HeaderXRequestIDKey = o.headerXRequestIDKey
}
}
......@@ -76,21 +76,21 @@ func RequestID(opts ...RequestIDOption) gin.HandlerFunc {
o.setRequestIDKey()
return func(c *gin.Context) {
requestID := c.Request.Header.Get(ctxUtil.HeaderXRequestIDKey.String())
requestID := c.Request.Header.Get(ctxUtil.HeaderXRequestIDKey)
// Create request id
if requestID == "" {
requestID = ctxUtil.GenerateTid()
c.Request.Header.Set(ctxUtil.HeaderXRequestIDKey.String(), requestID)
c.Request.Header.Set(ctxUtil.HeaderXRequestIDKey, requestID)
}
st := time.Now()
// Expose it for use in the application
c.Set(ctxUtil.ContextTraceIDKey.String(), requestID)
c.Set(ctxUtil.KeyApiStartTime.String(), st)
c.Set(ctxUtil.ContextTraceIDKey, requestID)
c.Set(ctxUtil.KeyApiStartTime, st)
// Set X-Request-Id header
c.Writer.Header().Set(ctxUtil.HeaderXRequestIDKey.String(), requestID)
c.Writer.Header().Set(ctxUtil.HeaderXRequestIDKey, requestID)
c.Next()
}
......@@ -98,12 +98,12 @@ func RequestID(opts ...RequestIDOption) gin.HandlerFunc {
// HeaderRequestID get request id from the header
func HeaderRequestID(c *gin.Context) string {
return c.Request.Header.Get(ctxUtil.HeaderXRequestIDKey.String())
return c.Request.Header.Get(ctxUtil.HeaderXRequestIDKey)
}
// HeaderRequestIDField get request id field from header
func HeaderRequestIDField(c *gin.Context) zap.Field {
return zap.String(ctxUtil.HeaderXRequestIDKey.String(), HeaderRequestID(c))
return zap.String(ctxUtil.HeaderXRequestIDKey, HeaderRequestID(c))
}
// -------------------------------------------------------------------------------------------
......@@ -113,7 +113,7 @@ var RequestHeaderKey = "request_header_key"
// WrapCtx wrap context, put the Keys and Header of gin.Context into context
func WrapCtx(c *gin.Context) context.Context {
ctx := context.WithValue(c.Request.Context(), ctxUtil.ContextTraceIDKey, c.GetString(ctxUtil.ContextTraceIDKey.String())) //nolint
ctx := context.WithValue(c.Request.Context(), ctxUtil.ContextTraceIDKey, c.GetString(ctxUtil.ContextTraceIDKey)) //nolint
for k, v := range c.Keys {
ctx = context.WithValue(ctx, k, v) //nolint
}
......@@ -122,7 +122,7 @@ func WrapCtx(c *gin.Context) context.Context {
}
func WrapAsyncCtx(c *gin.Context) context.Context {
ctx := context.WithValue(context.Background(), ctxUtil.ContextTraceIDKey, c.GetString(ctxUtil.ContextTraceIDKey.String())) //nolint
ctx := context.WithValue(context.Background(), ctxUtil.ContextTraceIDKey, c.GetString(ctxUtil.ContextTraceIDKey)) //nolint
for k, v := range c.Keys {
ctx = context.WithValue(ctx, k, v) //nolint
}
......
gin/middleware/requstid_test.go deleted 100644 → 0
浏览文件 @ eeaf1350
package middleware
import (
"context"
"net/http"
"testing"
"time"
ctxUtil "gitlab.wanzhuangkj.com/tush/xpkg/gin/ctxUtils"
"gitlab.wanzhuangkj.com/tush/xpkg/utils"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/assert"
)
func runRequestIDHTTPServer(fn func(c *gin.Context)) string {
serverAddr, requestAddr := utils.GetLocalHTTPAddrPairs()
gin.SetMode(gin.ReleaseMode)
r := gin.Default()
r.Use(RequestID(
//WithRequestIDKey("my_req_id"),
//WithHeaderRequestIDKey("My-X-Req-Id"),
))
r.GET("/ping", func(c *gin.Context) {
fn(c)
c.String(200, "pong")
})
go func() {
err := r.Run(serverAddr)
if err != nil {
panic(err)
}
}()
time.Sleep(time.Millisecond * 200)
return requestAddr
}
func TestFieldRequestIDFromContext(t *testing.T) {
requestAddr := runRequestIDHTTPServer(func(c *gin.Context) {
str := ctxUtil.CtxRequestID(c)
t.Log(str)
field := ctxUtil.CtxTraceIDField(c)
t.Log(field)
str = HeaderRequestID(c)
t.Log(str)
field = HeaderRequestIDField(c)
t.Log(field)
str = ctxUtil.CtxRequestID(c)
t.Log(str)
field = ctxUtil.CtxTraceIDField(c)
t.Log(field)
c.Set("foo", "bar")
ctx := WrapCtx(c)
t.Log(ctx.Value(ctxUtil.ContextTraceIDKey))
t.Log(GetFromCtx(ctx, "foo"))
t.Log(ctxUtil.CtxTraceIDField(ctx))
t.Log(GetFromCtx(ctx, "not-exist"))
t.Log(GetFromHeader(ctx, ctxUtil.HeaderXRequestIDKey.String()))
t.Log(GetFromHeader(ctx, "Accept"))
t.Log(GetFromHeader(ctx, "not-exist"))
t.Log(GetFromHeaders(ctx, "Accept"))
t.Log(GetFromHeaders(ctx, "not-exist"))
cctx := c
c2, ctx2 := AdaptCtx(cctx)
t.Log(ctxUtil.CtxRequestID(c2))
t.Log(ctx2.Value(ctxUtil.ContextTraceIDKey))
})
_, err := http.Get(requestAddr + "/ping")
assert.NoError(t, err)
defer func() { recover() }()
req, _ := http.NewRequest("GET", requestAddr+"/ping", nil)
req.Header.Add("Accept", "application/json")
req.Header.Add("Accept", "text/html")
req.Header.Set(ctxUtil.HeaderXRequestIDKey.String(), "2ab996de-cc03-412d-ba0a-79596efa6947")
resp, err := http.DefaultClient.Do(req)
assert.NoError(t, err)
defer resp.Body.Close()
}
func TestGetRequestIDFromContext(t *testing.T) {
str := ctxUtil.CtxRequestID(&gin.Context{})
assert.Equal(t, "", str)
str = ctxUtil.CtxRequestID(context.Background())
assert.Equal(t, "", str)
}
func TestRequestIDKeyOptions(t *testing.T) {
opts := []RequestIDOption{
WithContextRequestIDKey("xx"), // invalid settings
WithContextRequestIDKey("my_req_id"),
WithHeaderRequestIDKey("xx"), // invalid settings
WithHeaderRequestIDKey("My-X-Req-Id"),
}
o := defaultRequestIDOptions()
o.apply(opts...)
o.setRequestIDKey()
t.Log(ctxUtil.ContextTraceIDKey, ctxUtil.HeaderXRequestIDKey)
assert.Equal(t, "my_req_id", ctxUtil.ContextTraceIDKey)
assert.Equal(t, "My-X-Req-Id", ctxUtil.HeaderXRequestIDKey)
}
gin/response/response.go
浏览文件 @ ce21a10a
......@@ -143,9 +143,9 @@ func Out(c *gin.Context, err *errcode.Error, data ...interface{}) {
// status code flat 200, custom error codes in data.code
func respJSONWith200(c *gin.Context, code int, msg string, data ...interface{}) {
c.Writer.Header().Set(ctxUtils.HeaderXTimestampKey.String(), strconv.FormatInt(time.Now().Unix(), 10))
appName := c.GetString(ctxUtils.KeyAppName.String())
stTime := c.GetTime(ctxUtils.KeyApiStartTime.String())
c.Writer.Header().Set(ctxUtils.HeaderXTimestampKey, strconv.FormatInt(time.Now().Unix(), 10))
appName := c.GetString(ctxUtils.KeyAppName)
stTime := c.GetTime(ctxUtils.KeyApiStartTime)
ctxUtils.AddApiCost(c, appName, stTime)
if len(data) > 0 {
......@@ -158,13 +158,13 @@ func respJSONWith200(c *gin.Context, code int, msg string, data ...interface{})
// Success return success
func Success(c *gin.Context, data ...interface{}) {
c.Set(ctxUtils.KeyRspCode.String(), 1)
c.Set(ctxUtils.KeyRspCode, 1)
respJSONWith200(c, errcode.Success.Code(), errcode.Success.Msg(), data...)
}
// SuccessWithPage return success
func SuccessWithPage[T any](c *gin.Context, list []*T, total int64) {
c.Set(ctxUtils.KeyRspCode.String(), 1)
c.Set(ctxUtils.KeyRspCode, 1)
if list == nil {
list = []*T{}
}
......@@ -173,7 +173,7 @@ func SuccessWithPage[T any](c *gin.Context, list []*T, total int64) {
// SuccessWithList return success
func SuccessWithList[T any](c *gin.Context, list []*T) {
c.Set(ctxUtils.KeyRspCode.String(), 1)
c.Set(ctxUtils.KeyRspCode, 1)
if list == nil {
list = []*T{}
}
......@@ -182,12 +182,12 @@ func SuccessWithList[T any](c *gin.Context, list []*T) {
// ErrorE return error
func ErrorE(c *gin.Context, err *errcode.Error, data ...interface{}) {
c.Set(ctxUtils.KeyRspCode.String(), 0)
c.Set(ctxUtils.KeyRspCode, 0)
respJSONWith200(c, err.Code(), err.Msg(), data...)
}
func Error(c *gin.Context, err error) {
c.Set(ctxUtils.KeyRspCode.String(), 0)
c.Set(ctxUtils.KeyRspCode, 0)
msg := ""
if err != nil {
msg = err.Error()
......
httpcli/entity/entity.go
浏览文件 @ ce21a10a
......@@ -3,6 +3,7 @@ package entity
import (
"bytes"
"fmt"
"github.com/gin-gonic/gin"
"gitlab.wanzhuangkj.com/tush/xpkg/gin/ctxUtils"
)
......@@ -25,11 +26,11 @@ func SetCopyReq(c *gin.Context, buf *bytes.Buffer) {
copyHttpReq.URL = c.Request.URL.String()
copyHttpReq.Headers = c.Request.Header
copyHttpReq.Body = buf.String()
c.Set(ctxUtils.KeyReqBody.String(), copyHttpReq)
c.Set(ctxUtils.KeyReqBody, copyHttpReq)
}
func GetCopyReq(c *gin.Context) *CopyHttpReq {
if val, isExist := c.Get(ctxUtils.KeyReqBody.String()); isExist {
if val, isExist := c.Get(ctxUtils.KeyReqBody); isExist {
if req, ok := val.(*CopyHttpReq); ok {
return req
}
......@@ -41,11 +42,11 @@ func SetCopyRsp(c *gin.Context, buf *bytes.Buffer) {
copyHttpRsp := &CopyHttpRsp{}
copyHttpRsp.Headers = c.Writer.Header()
copyHttpRsp.Body = buf.String()
c.Set(ctxUtils.KeyRspBody.String(), copyHttpRsp)
c.Set(ctxUtils.KeyRspBody, copyHttpRsp)
}
func GetCopyRsp(c *gin.Context) *CopyHttpRsp {
if val, isExist := c.Get(ctxUtils.KeyRspBody.String()); isExist {
if val, isExist := c.Get(ctxUtils.KeyRspBody); isExist {
if rsp, ok := val.(*CopyHttpRsp); ok {
return rsp
}
......@@ -59,11 +60,11 @@ func SetCopyApiReq(c *gin.Context, buf *bytes.Buffer) {
copyHttpReq.URL = c.Request.URL.String()
copyHttpReq.Headers = c.Request.Header
copyHttpReq.Body = buf.String()
c.Set(fmt.Sprintf(ctxUtils.KeyOApiReqBody.String(), ctxUtils.GetGinCtxTraceID(c)), copyHttpReq)
c.Set(fmt.Sprintf(ctxUtils.KeyOApiReqBody, ctxUtils.GetGinCtxTraceID(c)), copyHttpReq)
}
func GetCopyApiReq(c *gin.Context) *CopyHttpReq {
if val, isExist := c.Get(fmt.Sprintf(ctxUtils.KeyOApiReqBody.String(), ctxUtils.GetGinCtxTraceID(c))); isExist {
if val, isExist := c.Get(fmt.Sprintf(ctxUtils.KeyOApiReqBody, ctxUtils.GetGinCtxTraceID(c))); isExist {
if req, ok := val.(*CopyHttpReq); ok {
return req
}
......@@ -75,11 +76,11 @@ func SetCopyApiRsp(c *gin.Context, buf *bytes.Buffer) {
copyHttpRsp := &CopyHttpRsp{}
copyHttpRsp.Headers = c.Writer.Header()
copyHttpRsp.Body = buf.String()
c.Set(fmt.Sprintf(ctxUtils.KeyOApiRspBody.String(), ctxUtils.GetGinCtxTraceID(c)), copyHttpRsp)
c.Set(fmt.Sprintf(ctxUtils.KeyOApiRspBody, ctxUtils.GetGinCtxTraceID(c)), copyHttpRsp)
}
func GetCopyApiRsp(c *gin.Context) *CopyHttpRsp {
if val, isExist := c.Get(fmt.Sprintf(ctxUtils.KeyOApiRspBody.String(), ctxUtils.GetGinCtxTraceID(c))); isExist {
if val, isExist := c.Get(fmt.Sprintf(ctxUtils.KeyOApiRspBody, ctxUtils.GetGinCtxTraceID(c))); isExist {
if rsp, ok := val.(*CopyHttpRsp); ok {
return rsp
}
......
httpcli/httpcli.go
浏览文件 @ ce21a10a
......@@ -290,7 +290,7 @@ func (x *Request) send(ctx context.Context) {
x.request.Header.Add(k, v)
}
}
x.request.Header.Add(ctxUtils.HeaderXRequestIDKey.String(), ctxUtils.CtxRequestID(ctx))
x.request.Header.Add(ctxUtils.HeaderXRequestIDKey, ctxUtils.CtxRequestID(ctx))
if x.timeout < 1 {
x.timeout = defaultTimeout
}
......@@ -378,7 +378,6 @@ func (x *Request) pushDo(ctx context.Context) {
var rsp *http.Response
rsp, x.err = client.Do(x.request)
if x.err != nil {
x.request = nil
return
}
if rsp != nil {
......@@ -390,7 +389,6 @@ func (x *Request) pushDo(ctx context.Context) {
defer rsp.Body.Close()
x.response.body, x.err = io.ReadAll(rsp.Body)
if x.err != nil {
x.request = nil
return
}
if len(x.response.body) > 0 {
......@@ -400,7 +398,6 @@ func (x *Request) pushDo(ctx context.Context) {
}
}
}
x.request = nil
}
// ----------------------------------- Response -----------------------------------
......
ws/ws.go
浏览文件 @ ce21a10a
......@@ -7,7 +7,6 @@ import (
"sync"
"time"
ctxUtil "gitlab.wanzhuangkj.com/tush/xpkg/gin/ctxUtils"
"gitlab.wanzhuangkj.com/tush/xpkg/jwt"
"gitlab.wanzhuangkj.com/tush/xpkg/logger"
"gitlab.wanzhuangkj.com/tush/xpkg/utils/xsf"
......@@ -27,60 +26,54 @@ const (
)
type wsConn struct {
manager *WsConnManager
id xsf.ID
conn *websocket.Conn
sendBuff chan []byte
log *zap.Logger
stop chan struct{}
manager *WsConnManager
receiverID xsf.ID // user id
conn *websocket.Conn
sendBuff chan *WsMessage
log *zap.Logger
stop chan struct{}
wg *sync.WaitGroup
}
type WsConnManager struct {
pool map[xsf.ID]*wsConn
upgrader *websocket.Upgrader
sync.RWMutex
log *zap.Logger
hooks []Hook
l *sync.RWMutex
log *zap.Logger
hooks []Hook
}
var (
WsConnMgr *WsConnManager
)
type Hook interface {
AfterSend(ctx context.Context, receiverID xsf.ID, payload *WsMessage) error
AfterAccept(ctx context.Context, userID xsf.ID) error
AfterUnregister(ctx context.Context, userID xsf.ID) error
AfterSend(receiverID xsf.ID, payload *WsMessage) error
AfterAccept(userID xsf.ID) error
AfterUnregister(userID xsf.ID) error
OnReceive(senderID xsf.ID, messageType int, msg []byte)
}
func NewWsConnManager(logger *zap.Logger, upgrader *websocket.Upgrader, hooks ...Hook) *WsConnManager {
return &WsConnManager{
pool: make(map[xsf.ID]*wsConn),
log: logger, upgrader: upgrader, hooks: hooks}
pool: make(map[xsf.ID]*wsConn),
log: logger,
upgrader: upgrader,
hooks: hooks,
l: &sync.RWMutex{},
}
}
func (m *WsConnManager) register(ctx context.Context, conn *wsConn) {
m.Lock()
defer m.Unlock()
if oldClient, exists := m.pool[conn.id]; exists {
oldClient.close(ctx)
func (m *WsConnManager) register(conn *wsConn) {
if oldClient, exists := m.pool[conn.receiverID]; exists {
oldClient.stop <- struct{}{}
}
m.pool[conn.id] = conn
m.pool[conn.receiverID] = conn
}
// 注销连接
func (conn *wsConn) close(ctx context.Context) {
conn.manager.Lock()
defer conn.manager.Unlock()
delete(conn.manager.pool, conn.id)
func (conn *wsConn) close() {
close(conn.sendBuff)
conn.stop <- struct{}{}
close(conn.stop)
conn.conn.Close()
for _, h := range conn.manager.hooks {
go func() {
h.AfterUnregister(ctx, conn.id)
}()
h.AfterUnregister(conn.receiverID)
}
}
......@@ -88,7 +81,7 @@ func (conn *wsConn) close(ctx context.Context) {
func (m *WsConnManager) Accept(c *gin.Context) {
err := m.accept(c)
if err != nil {
m.log.Error("handleWsConn failed", logger.Err(err))
m.log.Error("accept conn failed", logger.Err(err))
}
}
......@@ -114,44 +107,35 @@ func (m *WsConnManager) accept(c *gin.Context) error {
return xerror.Wrap(err, "ws upgrade failed")
}
client := &wsConn{
id: userID,
conn: conn,
sendBuff: make(chan []byte, 256),
manager: m,
log: m.log,
stop: make(chan struct{}),
receiverID: userID,
conn: conn,
sendBuff: make(chan *WsMessage, 256),
manager: m,
log: m.log,
stop: make(chan struct{}),
wg: &sync.WaitGroup{},
}
go client.start()
ctx := ctxUtil.WrapCtx(c)
m.register(ctx, client)
go client.heartBeat()
go client.writeLoop()
m.register(client)
for _, h := range m.hooks {
go func() {
h.AfterAccept(ctx, userID)
}()
h.AfterAccept(userID)
}
return nil
}
func (m *WsConnManager) getConn(userID xsf.ID) (*wsConn, bool) {
m.RLock()
defer m.RUnlock()
m.l.RLock()
defer m.l.RUnlock()
client, exists := m.pool[userID]
return client, exists
}
func (m *WsConnManager) Send(ctx context.Context, receiverID xsf.ID, payload *WsMessage) error {
if payload == nil {
return nil
}
conn, exists := m.getConn(receiverID)
if exists {
data, _ := json.Marshal(payload)
err := conn.send(ctx, data)
err := conn.send(ctx, payload)
if err != nil {
return err
}
......@@ -160,42 +144,117 @@ func (m *WsConnManager) Send(ctx context.Context, receiverID xsf.ID, payload *Ws
}
for _, h := range m.hooks {
go func() {
h.AfterSend(ctx, receiverID, payload)
h.AfterSend(receiverID, payload)
}()
}
return nil
}
func (c *wsConn) send(_ context.Context, msg []byte) error {
// todo 1. deadline time 2. retry
func (c *wsConn) start() {
c.wg.Add(3)
go c.heartBeat()
go c.writeLoop()
go c.readLoop()
c.wg.Wait()
c.close()
}
func (c *wsConn) send(_ context.Context, msg *WsMessage) error {
c.sendBuff <- msg
return nil
}
func (c *wsConn) heartBeat() {
c.log.Info("[ws][hearbeat]start")
ticker := time.NewTicker(heartbeatDuration)
defer ticker.Stop()
defer func() {
c.log.Info("[ws][hearbeat]stop")
ticker.Stop()
c.stop <- struct{}{}
c.wg.Done()
}()
for {
select {
case <-ticker.C:
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
defer c.close(context.Background())
c.log.Error("[ws][hearbeat]ping failed", logger.Err(err))
c.stop <- struct{}{}
return
}
case <-c.stop:
c.log.Info("[ws][hearbeat]stop signal received")
return
}
}
}
func (c *wsConn) writeLoop() {
c.log.Info("[ws][writeLoop]start")
defer func() {
c.log.Info("[ws][writeLoop]stop")
c.stop <- struct{}{}
c.wg.Done()
}()
for {
select {
case msg := <-c.sendBuff:
if err := c.conn.WriteMessage(websocket.BinaryMessage, msg); err != nil {
c.log.Error("write message failed", logger.Err(err))
case msg, ok := <-c.sendBuff:
if !ok {
c.log.Info("[ws][writeLoop]send channel closed, writer exit")
return
}
data, _ := json.Marshal(msg)
if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil {
c.log.Error("[ws][writeLoop]write message failed", logger.Err(err))
return
}
case <-c.stop:
c.log.Info("[ws][writeLoop]stop signal received")
return
}
}
}
type WsMsg struct {
senderID xsf.ID
msgType int
msg []byte
}
func (c *wsConn) readLoop() {
c.log.Info("[ws][readLoop]start")
msgChan := make(chan WsMsg)
defer func() {
c.log.Info("[ws][readLoop]exit")
close(msgChan)
c.stop <- struct{}{}
c.wg.Done()
}()
// 读协程
go func() {
for {
msgType, msg, err := c.conn.ReadMessage()
c.log.Info("[ws][readLoop]receive message", logger.String("msg", string(msg)))
if err != nil {
c.log.Error("[ws][readLoop]read message failed", logger.Err(err))
return
}
msgChan <- WsMsg{c.receiverID, msgType, msg}
}
}()
for {
select {
case m, ok := <-msgChan:
if !ok {
c.log.Info("[ws][readLoop]reader channel closed, exit")
return
}
for _, hook := range c.manager.hooks {
hook.OnReceive(m.senderID, m.msgType, m.msg)
}
case <-c.stop:
c.log.Info("[ws][readLoop]stop signal received")
return
}
}
......@@ -206,7 +265,7 @@ const (
)
type WsMessage struct {
Seq string `json:"seq"`
Cmd Cmd `json:"cmd"`
Data any `json:"payload"`
Seq string `json:"seq"`
Cmd Cmd `json:"cmd"`
Data json.RawMessage `json:"payload"` // 原始JSON，目前是models.ChatMessage
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论