diff --git a/constants/websocket.go b/constants/websocket.go index a9a6aa5d..34e28eb7 100644 --- a/constants/websocket.go +++ b/constants/websocket.go @@ -5,14 +5,19 @@ type websocket string // websocket消息类型 const ( //鉴权失败 - WEBSOCKET_UNAUTH = "unAuth" + WEBSOCKET_UNAUTH = "WEBSOCKET_UNAUTH" //ws连接成功 - WEBSOCKET_CONNECT_SUCCESS = "connect-success" - //心跳信息 - WEBSOCKET_HEARTBEAT = "heartbeat" + WEBSOCKET_CONNECT_SUCCESS = "WEBSOCKET_CONNECT_SUCCESS" //图片渲染 - WEBSOCKET_RENDER_IMAGE = "render-image" + WEBSOCKET_RENDER_IMAGE = "WEBSOCKET_RENDER_IMAGE" + //数据格式错误 + WEBSOCKET_ERR_DATA_FORMAT = "WEBSOCKET_ERR_DATA_FORMAT" + //第三方登录通知 + WEBSOCKET_THIRD_PARTY_LOGIN_NOTIFY = "WEBSOCKET_THIRD_PARTY_LOGIN_NOTIFY" ) -// 云渲染通知需要的签名字符串 +// 云渲染完成通知api需要的签名字符串 const RENDER_NOTIFY_SIGN_KEY = "fusen-render-notify-%s-%d" + +// 第三方登录通知api需要的签名字符串 +const THIRD_PARTY_LOGIN_NOTIFY_SIGN_KEY = "fusen-render-notify-%s-%d" diff --git a/model/gmodel/fs_pay_gen.go b/model/gmodel/fs_pay_gen.go index 0ee65802..e30685b6 100644 --- a/model/gmodel/fs_pay_gen.go +++ b/model/gmodel/fs_pay_gen.go @@ -7,6 +7,7 @@ import ( // fs_pay 支付记录 type FsPay struct { Id int64 `gorm:"primary_key;default:0;auto_increment;" json:"id"` // + PayNo *string `gorm:"default:'';" json:"pay_no"` // 支付编号 UserId *int64 `gorm:"index;default:0;" json:"user_id"` // 用户id OrderNumber *string `gorm:"default:'';" json:"order_number"` // 订单编号 TradeNo *string `gorm:"index;default:'';" json:"trade_no"` // 第三方支付编号 diff --git a/model/gmodel/var_gen.go b/model/gmodel/var_gen.go index 191a9898..741c6b05 100644 --- a/model/gmodel/var_gen.go +++ b/model/gmodel/var_gen.go @@ -79,7 +79,10 @@ type AllModelsGen struct { FsQuotationRemarkTemplate *FsQuotationRemarkTemplateModel // fs_quotation_remark_template 报价单备注模板 FsQuotationSaler *FsQuotationSalerModel // fs_quotation_saler 报价单业务员表 FsRefundReason *FsRefundReasonModel // fs_refund_reason +<<<<<<< HEAD FsResource *FsResourceModel // fs_resource 资源表 +======= +>>>>>>> develop FsResources *FsResourcesModel // fs_resources 资源表 FsStandardLogo *FsStandardLogoModel // fs_standard_logo 标准logo FsTags *FsTagsModel // fs_tags 产品分类表 @@ -171,7 +174,10 @@ func NewAllModels(gdb *gorm.DB) *AllModelsGen { FsQuotationRemarkTemplate: NewFsQuotationRemarkTemplateModel(gdb), FsQuotationSaler: NewFsQuotationSalerModel(gdb), FsRefundReason: NewFsRefundReasonModel(gdb), +<<<<<<< HEAD FsResource: NewFsResourceModel(gdb), +======= +>>>>>>> develop FsResources: NewFsResourcesModel(gdb), FsStandardLogo: NewFsStandardLogoModel(gdb), FsTags: NewFsTagsModel(gdb), diff --git a/server/product-template-tag/internal/handler/routes.go b/server/product-template-tag/internal/handler/routes.go index b2c8f181..24a5c707 100644 --- a/server/product-template-tag/internal/handler/routes.go +++ b/server/product-template-tag/internal/handler/routes.go @@ -14,7 +14,7 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { []rest.Route{ { Method: http.MethodGet, - Path: "/api/product-template/get_product_template_tags", + Path: "/api/product-template-tag/get_product_template_tags", Handler: GetProductTemplateTagsHandler(serverCtx), }, }, diff --git a/server/product-template-tag/internal/logic/getproducttemplatetagslogic.go b/server/product-template-tag/internal/logic/getproducttemplatetagslogic.go index 68c62363..815ef07c 100644 --- a/server/product-template-tag/internal/logic/getproducttemplatetagslogic.go +++ b/server/product-template-tag/internal/logic/getproducttemplatetagslogic.go @@ -47,6 +47,7 @@ func (l *GetProductTemplateTagsLogic) GetProductTemplateTags(req *types.GetProdu list := make([]types.GetProductTemplateTagsRsp, 0, len(productTemplateTags)) for _, v := range productTemplateTags { list = append(list, types.GetProductTemplateTagsRsp{ + Id: v.Id, Tag: *v.Title, Cover: *v.CoverImg, }) diff --git a/server/product-template-tag/internal/types/types.go b/server/product-template-tag/internal/types/types.go index 39866da6..a89ea882 100644 --- a/server/product-template-tag/internal/types/types.go +++ b/server/product-template-tag/internal/types/types.go @@ -10,6 +10,7 @@ type GetProductTemplateTagsReq struct { } type GetProductTemplateTagsRsp struct { + Id int64 `json:"id"` Tag string `json:"tag"` Cover string `json:"cover"` } diff --git a/server/websocket/internal/handler/datatransferhandler.go b/server/websocket/internal/handler/datatransferhandler.go index 3e5177d7..8ef09842 100644 --- a/server/websocket/internal/handler/datatransferhandler.go +++ b/server/websocket/internal/handler/datatransferhandler.go @@ -1,270 +1,15 @@ package handler import ( - "encoding/json" - "fmt" - "fusenapi/constants" "fusenapi/server/websocket/internal/logic" "fusenapi/server/websocket/internal/svc" - "fusenapi/server/websocket/internal/types" - "fusenapi/utils/auth" - "github.com/google/uuid" - "github.com/gorilla/websocket" - "github.com/zeromicro/go-zero/core/logx" "net/http" - "sync" - "time" ) -var ( - //升级 - upgrade = websocket.Upgrader{ - //允许跨域 - CheckOrigin: func(r *http.Request) bool { - return true - }, - } - //连接map池 - mapConnPool = sync.Map{} -) - -// 每个连接的连接属性 -type wsConnectItem struct { - conn *websocket.Conn //websocket的连接 - closeChan chan struct{} //关闭chan - isClose bool //是否已经关闭 - flag string - inChan chan []byte //接受消息缓冲通道 - outChan chan []byte //发送回客户端的消息 - mutex sync.Mutex - renderImage map[string]struct{} //需要渲染的图片 -} - func DataTransferHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { // 创建一个业务逻辑层实例 - var req types.DataTransferReq l := logic.NewDataTransferLogic(r.Context(), svcCtx) - l.DataTransfer(&req, nil) - //升级websocket - conn, err := upgrade.Upgrade(w, r, nil) - if err != nil { - logx.Error("http upgrade websocket err:", err) - return - } - defer conn.Close() - rsp := types.DataTransferRsp{} - isAuth := true - // 解析JWT token,并对空用户进行判断 - claims, err := svcCtx.ParseJwtToken(r) - // 如果解析JWT token出错,则返回未授权的JSON响应并记录错误消息 - if err != nil { - rsp.T = constants.WEBSOCKET_UNAUTH - rsp.D = "unAuth" - b, _ := json.Marshal(rsp) - _ = conn.WriteMessage(websocket.TextMessage, b) - isAuth = false - } - if claims != nil { - // 从token中获取对应的用户信息 - _, err = auth.GetUserInfoFormMapClaims(claims) - // 如果获取用户信息出错,则返回未授权的JSON响应并记录错误消息 - if err != nil { - rsp.T = constants.WEBSOCKET_UNAUTH - rsp.D = "unAuth!!" - b, _ := json.Marshal(rsp) - _ = conn.WriteMessage(websocket.TextMessage, b) - isAuth = false - } - } else { - // 如果claims为nil,则认为用户身份为白板用户 - rsp.T = constants.WEBSOCKET_UNAUTH - rsp.D = "unAuth!!!" - b, _ := json.Marshal(rsp) - _ = conn.WriteMessage(websocket.TextMessage, b) - isAuth = false - } - //不是授权的连接(10秒后关闭) - if !isAuth { - select { - case <-time.After(time.Second * 10): - conn.Close() - return - } - } - //生成连接唯一标识 - flag := uuid.New().String() + time.Now().Format("20060102150405") - ws := wsConnectItem{ - conn: conn, - flag: flag, - closeChan: make(chan struct{}, 1), - inChan: make(chan []byte, 100), - outChan: make(chan []byte, 100), - renderImage: make(map[string]struct{}), - isClose: false, - } - //保存连接 - mapConnPool.Store(flag, ws) - defer ws.close() - //把连接成功消息发回去 - rsp.T = constants.WEBSOCKET_CONNECT_SUCCESS - rsp.D = flag - b, _ := json.Marshal(rsp) - conn.WriteMessage(websocket.TextMessage, b) - //循环读客户端信息 - go ws.readLoop() - //循环把数据发送给客户端 - go ws.writeLoop() - //推消息到云渲染 - go ws.sendLoop() - //心跳 - ws.heartbeat() - + l.DataTransfer(svcCtx, w, r) } } - -// 心跳 -func (w *wsConnectItem) heartbeat() { - rsp := types.DataTransferRsp{ - T: constants.WEBSOCKET_HEARTBEAT, - D: "", - } - b, _ := json.Marshal(rsp) - for { - time.Sleep(time.Second * 10) - select { - case <-w.closeChan: - return - default: - - } - //发送心跳信息 - if err := w.conn.WriteMessage(websocket.TextMessage, b); err != nil { - logx.Error("发送心跳信息异常，关闭连接:", w.flag, err) - w.close() - return - } - } -} - -// 关闭连接 -func (w *wsConnectItem) close() { - w.mutex.Lock() - defer w.mutex.Unlock() - logx.Info("websocket:", w.flag, " is closing...") - w.conn.Close() - mapConnPool.Delete(w.flag) - if !w.isClose { - w.isClose = true - close(w.closeChan) - close(w.outChan) - close(w.inChan) - } - logx.Info("websocket:", w.flag, " is closed") -} - -func (w *wsConnectItem) writeLoop() { - for { - select { - case <-w.closeChan: //如果关闭了 - return - case data := <-w.outChan: - w.conn.WriteMessage(websocket.TextMessage, data) - } - } -} - -// 接受客户端发来的消息 -func (w *wsConnectItem) readLoop() { - for { - select { - case <-w.closeChan: //如果关闭了 - return - default: - - } - _, data, err := w.conn.ReadMessage() - if err != nil { - logx.Error("接受信息错误:", err) - //关闭连接 - w.close() - return - } - //消息传入缓冲通道 - w.inChan <- data - } -} - -// 把收到的消息发往不同的地方处理 -func (w *wsConnectItem) sendLoop() { - for { - select { - case <-w.closeChan: - return - case data := <-w.inChan: - w.dealwithReciveData(data) - } - } -} - -// 处理接受到的数据 -func (w *wsConnectItem) dealwithReciveData(data []byte) { - var parseInfo types.DataTransferReq - if err := json.Unmarshal(data, &parseInfo); err != nil { - logx.Error("invalid format of websocket message") - return - } - switch parseInfo.T { - //图片渲染 - case constants.WEBSOCKET_RENDER_IMAGE: - var renderImageData []types.RenderImageReqMsg - if err := json.Unmarshal([]byte(parseInfo.D), &renderImageData); err != nil { - logx.Error("invalid format of websocket render image message", err) - return - } - logx.Info("收到请求云渲染图片数据：", renderImageData) - w.mutex.Lock() - defer w.mutex.Unlock() - //把需要渲染的图片加进去 - for _, v := range renderImageData { - key := w.getRenderImageMapKey(v.ProductId, v.SizeId, v.TemplateId) - w.renderImage[key] = struct{}{} - } - default: - - } -} - -// 把渲染好的数据放入outchan -func (w *wsConnectItem) setOutRenderImage(req types.RenderNotifyReq) { - w.mutex.Lock() - defer w.mutex.Unlock() - for _, notifyItem := range req.NotifyList { - renderKey := w.getRenderImageMapKey(notifyItem.ProductId, notifyItem.SizeId, notifyItem.TemplateId) - //查询 - _, ok := w.renderImage[renderKey] - if !ok { - continue - } - responseData := types.RenderImageRspMsg{ - ProductId: notifyItem.ProductId, - SizeId: notifyItem.SizeId, - TemplateId: notifyItem.TemplateId, - Source: "我是渲染资源", - } - b, _ := json.Marshal(responseData) - select { - case <-w.closeChan: - return - case w.outChan <- b: - logx.Info("notify send render result to out chan") - } - //删掉已经处理的渲染任务 - delete(w.renderImage, renderKey) - } -} - -// 获取需要渲染图片的map key -func (w *wsConnectItem) getRenderImageMapKey(productId, sizeId, templateId int64) string { - return fmt.Sprintf("%d-%d-%d", productId, sizeId, templateId) -} diff --git a/server/websocket/internal/handler/rendernotifyhandler.go b/server/websocket/internal/handler/rendernotifyhandler.go index 005ee044..ab16cdad 100644 --- a/server/websocket/internal/handler/rendernotifyhandler.go +++ b/server/websocket/internal/handler/rendernotifyhandler.go @@ -1,18 +1,12 @@ package handler import ( - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "fusenapi/constants" - "fusenapi/utils/basic" - "github.com/zeromicro/go-zero/rest/httpx" - "net/http" - "time" - + "fusenapi/server/websocket/internal/logic" "fusenapi/server/websocket/internal/svc" "fusenapi/server/websocket/internal/types" + "fusenapi/utils/basic" + "net/http" + "reflect" ) func RenderNotifyHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { @@ -20,56 +14,18 @@ func RenderNotifyHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { var req types.RenderNotifyReq _, err := basic.RequestParse(w, r, svcCtx, &req) if err != nil { - httpx.OkJsonCtx(r.Context(), w, basic.Response{ - Code: basic.CodeRequestParamsErr.Code, - Message: "err param", - Data: nil, - }) return } - if len(req.NotifyList) == 0 { - httpx.OkJsonCtx(r.Context(), w, basic.Response{ - Code: basic.CodeRequestParamsErr.Code, - Message: "invalid param,notify list is empty", - Data: nil, - }) - return + // 创建一个业务逻辑层实例 + l := logic.NewRenderNotifyLogic(r.Context(), svcCtx) + + rl := reflect.ValueOf(l) + basic.BeforeLogic(w, r, rl) + + resp := l.RenderNotify(&req) + + if !basic.AfterLogic(w, r, rl, resp) { + basic.NormalAfterLogic(w, r, resp) } - if time.Now().Unix()-120 > req.Time /*|| req.Time > time.Now().Unix() */ { - httpx.OkJsonCtx(r.Context(), w, basic.Response{ - Code: basic.CodeRequestParamsErr.Code, - Message: "invalid param,time is expired", - Data: nil, - }) - return - } - //验证签名 sha256 - notifyByte, _ := json.Marshal(req.NotifyList) - h := sha256.New() - h.Write([]byte(fmt.Sprintf(constants.RENDER_NOTIFY_SIGN_KEY, string(notifyByte), req.Time))) - signHex := h.Sum(nil) - sign := hex.EncodeToString(signHex) - if req.Sign != sign { - httpx.OkJsonCtx(r.Context(), w, basic.Response{ - Code: basic.CodeRequestParamsErr.Code, - Message: "invalid sign", - Data: nil, - }) - return - } - //遍历链接 - mapConnPool.Range(func(key, value any) bool { - ws, ok := value.(wsConnectItem) - if !ok { - return false - } - ws.setOutRenderImage(req) - return true - }) - httpx.OkJsonCtx(r.Context(), w, basic.Response{ - Code: 200, - Message: "success", - Data: nil, - }) } } diff --git a/server/websocket/internal/handler/routes.go b/server/websocket/internal/handler/routes.go index 859ecc5b..6649e3c3 100644 --- a/server/websocket/internal/handler/routes.go +++ b/server/websocket/internal/handler/routes.go @@ -22,6 +22,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { Path: "/api/websocket/render_notify", Handler: RenderNotifyHandler(serverCtx), }, + { + Method: http.MethodPost, + Path: "/api/websocket/third_party_login_notify", + Handler: ThirdPartyLoginNotifyHandler(serverCtx), + }, }, ) } diff --git a/server/websocket/internal/handler/thirdpartyloginnotifyhandler.go b/server/websocket/internal/handler/thirdpartyloginnotifyhandler.go new file mode 100644 index 00000000..ae16b24c --- /dev/null +++ b/server/websocket/internal/handler/thirdpartyloginnotifyhandler.go @@ -0,0 +1,35 @@ +package handler + +import ( + "net/http" + "reflect" + + "fusenapi/utils/basic" + + "fusenapi/server/websocket/internal/logic" + "fusenapi/server/websocket/internal/svc" + "fusenapi/server/websocket/internal/types" +) + +func ThirdPartyLoginNotifyHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + + var req types.ThirdPartyLoginNotifyReq + userinfo, err := basic.RequestParse(w, r, svcCtx, &req) + if err != nil { + return + } + + // 创建一个业务逻辑层实例 + l := logic.NewThirdPartyLoginNotifyLogic(r.Context(), svcCtx) + + rl := reflect.ValueOf(l) + basic.BeforeLogic(w, r, rl) + + resp := l.ThirdPartyLoginNotify(&req, userinfo) + + if !basic.AfterLogic(w, r, rl, resp) { + basic.NormalAfterLogic(w, r, resp) + } + } +} diff --git a/server/websocket/internal/logic/datatransferlogic.go b/server/websocket/internal/logic/datatransferlogic.go index 7b5db86b..44ba0abb 100644 --- a/server/websocket/internal/logic/datatransferlogic.go +++ b/server/websocket/internal/logic/datatransferlogic.go @@ -1,14 +1,21 @@ package logic import ( + "bytes" + "encoding/json" + "fmt" + "fusenapi/constants" + "fusenapi/server/websocket/internal/types" "fusenapi/utils/auth" - "fusenapi/utils/basic" + "fusenapi/utils/id_generator" + "github.com/gorilla/websocket" + "net/http" + "sync" + "time" "context" "fusenapi/server/websocket/internal/svc" - "fusenapi/server/websocket/internal/types" - "github.com/zeromicro/go-zero/core/logx" ) @@ -26,18 +33,247 @@ func NewDataTransferLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Data } } -// 处理进入前逻辑w,r -// func (l *DataTransferLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) { -// } +var ( + //全局websocketid生成器 + websocketIdGenerator = id_generator.NewWebsocketId(1) + //临时缓存对象池 + buffPool = sync.Pool{ + New: func() interface{} { + return bytes.Buffer{} + }, + } + //升级websocket + upgrade = websocket.Upgrader{ + //最大可读取大小 10M + ReadBufferSize: 1024 * 10, + //握手超时时间15s + HandshakeTimeout: time.Second * 15, + //允许跨域 + CheckOrigin: func(r *http.Request) bool { + return true + }, + //写的缓存池 + WriteBufferPool: &buffPool, + //是否支持压缩 + EnableCompression: true, + } + //websocket连接存储 + mapConnPool = sync.Map{} +) -// 处理逻辑后 w,r 如:重定向, resp 必须重新处理 -// func (l *DataTransferLogic) AfterLogic(w http.ResponseWriter, r *http.Request, resp *basic.Response) { -// // httpx.OkJsonCtx(r.Context(), w, resp) -// } - -func (l *DataTransferLogic) DataTransfer(req *types.DataTransferReq, userinfo *auth.UserInfo) (resp *basic.Response) { - // 返回值必须调用Set重新返回, resp可以空指针调用 resp.SetStatus(basic.CodeOK, data) - // userinfo 传入值时, 一定不为null - - return resp.SetStatus(basic.CodeOK) +// 每个连接的连接基本属性 +type wsConnectItem struct { + conn *websocket.Conn //websocket的连接 + closeChan chan struct{} //ws连接关闭chan + isClose bool //是否已经关闭 + uniqueId uint64 //ws连接唯一标识 + inChan chan []byte //接受消息缓冲通道 + outChan chan []byte //发送回客户端的消息 + mutex sync.Mutex //互斥锁 + renderProperty renderProperty //扩展云渲染属性 +} + +func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.ResponseWriter, r *http.Request) { + //升级websocket + conn, err := upgrade.Upgrade(w, r, nil) + if err != nil { + logx.Error("http upgrade websocket err:", err) + return + } + defer conn.Close() + //鉴权不成功10秒后断开 + /*isAuth, _ := l.checkAuth(svcCtx, r) + if !isAuth { + time.Sleep(time.Second) //兼容下火狐 + rsp := types.DataTransferData{ + T: constants.WEBSOCKET_UNAUTH, + D: nil, + } + b, _ := json.Marshal(rsp) + //先发一条正常信息 + _ = conn.WriteMessage(websocket.TextMessage, b) + //发送关闭信息 + _ = conn.WriteMessage(websocket.CloseMessage, nil) + return + }*/ + //生成连接唯一标识 + uniqueId := websocketIdGenerator.Get() + ws := wsConnectItem{ + conn: conn, + uniqueId: uniqueId, + closeChan: make(chan struct{}, 1), + inChan: make(chan []byte, 1000), + outChan: make(chan []byte, 1000), + renderProperty: renderProperty{ + renderImageTask: make(map[string]struct{}), + renderImageTaskCtlChan: make(chan renderImageControlChanItem, 100), + }, + } + //保存连接 + mapConnPool.Store(uniqueId, ws) + defer ws.close() + //把连接成功消息发回去 + time.Sleep(time.Second) //兼容下火狐 + b := ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, uniqueId) + _ = conn.WriteMessage(websocket.TextMessage, b) + //循环读客户端信息 + go ws.readLoop() + //循环把数据发送给客户端 + go ws.writeLoop() + //推消息到云渲染 + go ws.sendLoop() + //操作连接中渲染任务的增加/删除 + go ws.operationRenderTask() + //心跳 + ws.heartbeat() +} + +// 鉴权 +func (l *DataTransferLogic) checkAuth(svcCtx *svc.ServiceContext, r *http.Request) (isAuth bool, userInfo *auth.UserInfo) { + // 解析JWT token,并对空用户进行判断 + claims, err := svcCtx.ParseJwtToken(r) + // 如果解析JWT token出错,则返回未授权的JSON响应并记录错误消息 + if err != nil { + return false, nil + } + if claims != nil { + // 从token中获取对应的用户信息 + userInfo, err = auth.GetUserInfoFormMapClaims(claims) + // 如果获取用户信息出错,则返回未授权的JSON响应并记录错误消息 + if err != nil { + return false, nil + } + } else { + return false, nil + } + return true, userInfo +} + +// 心跳 +func (w *wsConnectItem) heartbeat() { + tick := time.Tick(time.Second * 5) + for { + select { + case <-w.closeChan: + return + case <-tick: + //发送心跳信息 + if err := w.conn.WriteMessage(websocket.PongMessage, nil); err != nil { + logx.Error("发送心跳信息异常，关闭连接:", w.uniqueId, err) + w.close() + return + } + } + } +} + +// 关闭连接 +func (w *wsConnectItem) close() { + w.mutex.Lock() + defer w.mutex.Unlock() + logx.Info("websocket:", w.uniqueId, " is closing...") + //发送关闭信息 + _ = w.conn.WriteMessage(websocket.CloseMessage, nil) + w.conn.Close() + mapConnPool.Delete(w.uniqueId) + if !w.isClose { + w.isClose = true + close(w.closeChan) + } + logx.Info("websocket:", w.uniqueId, " is closed") +} + +// 读取输出返回给客户端 +func (w *wsConnectItem) writeLoop() { + for { + select { + case <-w.closeChan: //如果关闭了 + return + case data := <-w.outChan: + if err := w.conn.WriteMessage(websocket.TextMessage, data); err != nil { + logx.Error("websocket write loop err:", err) + w.close() + return + } + } + } +} + +// 接受客户端发来的消息 +func (w *wsConnectItem) readLoop() { + for { + select { + case <-w.closeChan: //如果关闭了 + return + default: + msgType, data, err := w.conn.ReadMessage() + if err != nil { + logx.Error("接受信息错误:", err) + //关闭连接 + w.close() + return + } + //ping的消息不处理 + if msgType != websocket.PingMessage { + //消息传入缓冲通道 + w.inChan <- data + } + } + } +} + +// 把收到的消息发往不同的地方处理 +func (w *wsConnectItem) sendLoop() { + for { + select { + case <-w.closeChan: + return + case data := <-w.inChan: + w.dealwithReciveData(data) + } + } +} + +// 把要传递给客户端的数据放入outchan +func (w *wsConnectItem) sendToOutChan(data []byte) { + select { + case <-w.closeChan: + return + case w.outChan <- data: + logx.Info("notify send render result to out chan") + } +} + +// 获取需要渲染图片的map key +func (w *wsConnectItem) getRenderImageMapKey(productId, templateTagId, logoId int64, algorithmVersion string) string { + return fmt.Sprintf("%d-%d-%d-%s", productId, templateTagId, logoId, algorithmVersion) +} + +// 格式化返回数据 +func (w *wsConnectItem) respondDataFormat(msgType string, data interface{}) []byte { + d := types.DataTransferData{ + T: msgType, + D: data, + } + b, _ := json.Marshal(d) + return b +} + +// 处理接受到的数据 +func (w *wsConnectItem) dealwithReciveData(data []byte) { + var parseInfo types.DataTransferData + if err := json.Unmarshal(data, &parseInfo); err != nil { + logx.Error("invalid format of websocket message") + w.outChan <- w.respondDataFormat(constants.WEBSOCKET_ERR_DATA_FORMAT, "invalid format of websocket message:"+string(data)) + return + } + d, _ := json.Marshal(parseInfo.D) + //分消息类型给到不同逻辑处理，可扩展 + switch parseInfo.T { + //图片渲染 + case constants.WEBSOCKET_RENDER_IMAGE: + go w.SendToCloudRender(d) + default: + + } } diff --git a/server/websocket/internal/logic/rendernotifylogic.go b/server/websocket/internal/logic/rendernotifylogic.go index 49803d29..9e5e0c9e 100644 --- a/server/websocket/internal/logic/rendernotifylogic.go +++ b/server/websocket/internal/logic/rendernotifylogic.go @@ -1,8 +1,9 @@ package logic import ( - "fusenapi/utils/auth" + "fusenapi/constants" "fusenapi/utils/basic" + "time" "context" @@ -35,9 +36,50 @@ func NewRenderNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Rend // // httpx.OkJsonCtx(r.Context(), w, resp) // } -func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) { - // 返回值必须调用Set重新返回, resp可以空指针调用 resp.SetStatus(basic.CodeOK, data) - // userinfo 传入值时, 一定不为null - +func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq) (resp *basic.Response) { + if time.Now().Unix()-120 > req.Time /*|| req.Time > time.Now().Unix() */ { + return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param time") + } + //验证签名 sha256 + /*notifyByte, _ := json.Marshal(req.Info) + h := sha256.New() + h.Write([]byte(fmt.Sprintf(constants.RENDER_NOTIFY_SIGN_KEY, string(notifyByte), req.Time))) + signHex := h.Sum(nil) + sign := hex.EncodeToString(signHex) + //fmt.Println(sign) + if req.Sign != sign { + return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid sign") + }*/ + //遍历websocket链接把数据传进去 + mapConnPool.Range(func(key, value any) bool { + //断言连接 + ws, ok := value.(wsConnectItem) + if !ok { + return true + } + //关闭标识 + if ws.isClose { + return true + } + renderKey := ws.getRenderImageMapKey(req.Info.ProductId, req.Info.TemplateTagId, req.Info.LogoId, req.Info.AlgorithmVersion) + //查询有无该渲染任务 + _, ok = ws.renderProperty.renderImageTask[renderKey] + if !ok { + return true + } + b := ws.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, types.RenderImageRspMsg{ + ProductId: req.Info.ProductId, + TemplateTagId: req.Info.TemplateTagId, + Image: req.Info.Image, + }) + //删除对应的需要渲染的图片map + ws.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{ + Option: 0, //0删除 1添加 + Key: renderKey, + } + //发送数据到out chan + ws.sendToOutChan(b) + return true + }) return resp.SetStatus(basic.CodeOK) } diff --git a/server/websocket/internal/logic/thirdpartyloginnotifylogic.go b/server/websocket/internal/logic/thirdpartyloginnotifylogic.go new file mode 100644 index 00000000..d363ffd9 --- /dev/null +++ b/server/websocket/internal/logic/thirdpartyloginnotifylogic.go @@ -0,0 +1,78 @@ +package logic + +import ( + "fusenapi/constants" + "fusenapi/utils/auth" + "fusenapi/utils/basic" + "time" + + "context" + + "fusenapi/server/websocket/internal/svc" + "fusenapi/server/websocket/internal/types" + + "github.com/zeromicro/go-zero/core/logx" +) + +type ThirdPartyLoginNotifyLogic struct { + logx.Logger + ctx context.Context + svcCtx *svc.ServiceContext +} + +func NewThirdPartyLoginNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ThirdPartyLoginNotifyLogic { + return &ThirdPartyLoginNotifyLogic{ + Logger: logx.WithContext(ctx), + ctx: ctx, + svcCtx: svcCtx, + } +} + +// 处理进入前逻辑w,r +// func (l *ThirdPartyLoginNotifyLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) { +// } + +// 处理逻辑后 w,r 如:重定向, resp 必须重新处理 +// func (l *ThirdPartyLoginNotifyLogic) AfterLogic(w http.ResponseWriter, r *http.Request, resp *basic.Response) { +// // httpx.OkJsonCtx(r.Context(), w, resp) +// } + +func (l *ThirdPartyLoginNotifyLogic) ThirdPartyLoginNotify(req *types.ThirdPartyLoginNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) { + if time.Now().Unix()-120 > req.Time /*|| req.Time > time.Now().Unix() */ { + return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "err param: time is invalid") + } + if req.Info.WebsocketId <= 0 { + return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "err param:websocket_id is required") + } + if req.Info.Token == "" { + return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "err param:token is required") + } + //验证签名 sha256 + /*notifyByte, _ := json.Marshal(req.Info) + h := sha256.New() + h.Write([]byte(fmt.Sprintf(constants.THIRD_PARTY_LOGIN_NOTIFY_SIGN_KEY, string(notifyByte), req.Time))) + signHex := h.Sum(nil) + sign := hex.EncodeToString(signHex) + //fmt.Println(sign) + if req.Sign != sign { + return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid sign") + }*/ + //查询对应websocket连接 + val, ok := mapConnPool.Load(req.Info.WebsocketId) + if !ok { + return resp.SetStatusWithMessage(basic.CodeOK, "success:websocket connection is not exists") + } + ws, ok := val.(wsConnectItem) + if !ok { + return resp.SetStatusWithMessage(basic.CodeServiceErr, "type of websocket connect object is err") + } + b := ws.respondDataFormat(constants.WEBSOCKET_THIRD_PARTY_LOGIN_NOTIFY, types.ThirdPartyLoginRspMsg{ + Token: req.Info.Token, + }) + select { + case <-ws.closeChan: + return resp.SetStatusWithMessage(basic.CodeOK, "websocket connect object is closed") + case ws.outChan <- b: + return resp.SetStatusWithMessage(basic.CodeOK, "success") + } +} diff --git a/server/websocket/internal/logic/ws_render_image_logic.go b/server/websocket/internal/logic/ws_render_image_logic.go new file mode 100644 index 00000000..d58def77 --- /dev/null +++ b/server/websocket/internal/logic/ws_render_image_logic.go @@ -0,0 +1,65 @@ +package logic + +import ( + "encoding/json" + "fusenapi/constants" + "fusenapi/server/websocket/internal/types" + "github.com/zeromicro/go-zero/core/logx" +) + +// 云渲染属性 +type renderProperty struct { + renderImageTask map[string]struct{} //需要渲染的图片任务 + renderImageTaskCtlChan chan renderImageControlChanItem //渲染任务新增移除的控制通道 +} + +// 渲染任务新增移除的控制通道的数据 +type renderImageControlChanItem struct { + Option int // 0删除 1添加 + Key string //map的key +} + +// 操作连接中渲染任务的增加/删除 +func (w *wsConnectItem) operationRenderTask() { + for { + select { + case <-w.closeChan: + return + case data := <-w.renderProperty.renderImageTaskCtlChan: + switch data.Option { + case 0: //删除任务 + delete(w.renderProperty.renderImageTask, data.Key) + case 1: //新增任务 + w.renderProperty.renderImageTask[data.Key] = struct{}{} + default: + + } + } + } +} + +// 渲染请求数据处理发送云渲染服务处理 +func (w *wsConnectItem) SendToCloudRender(data []byte) { + var renderImageData types.RenderImageReqMsg + if err := json.Unmarshal(data, &renderImageData); err != nil { + w.outChan <- w.respondDataFormat(constants.WEBSOCKET_ERR_DATA_FORMAT, "invalid format of websocket render image message:"+string(data)) + logx.Error("invalid format of websocket render image message", err) + return + } + logx.Info("收到请求云渲染图片数据：", renderImageData) + //把需要渲染的图片任务加进去 + for _, productId := range renderImageData.ProductIds { + select { + case <-w.closeChan: //连接关闭了 + return + default: + //加入渲染任务 + key := w.getRenderImageMapKey(productId, renderImageData.TemplateTagId, renderImageData.LogoId, renderImageData.AlgorithmVersion) + w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{ + Option: 1, //0删除 1添加 + Key: key, + } + // TODO 数据发送给云渲染服务器 + } + } +} diff --git a/server/websocket/internal/types/types.go b/server/websocket/internal/types/types.go index 6a1cf750..901b8123 100644 --- a/server/websocket/internal/types/types.go +++ b/server/websocket/internal/types/types.go @@ -5,40 +5,53 @@ import ( "fusenapi/utils/basic" ) -type DataTransferReq struct { - T string `json:"t"` //消息类型 - D string `json:"d"` //传递的消息 -} - -type DataTransferRsp struct { - T string `json:"t"` //消息类型 - D string `json:"d"` //传递的消息 +type DataTransferData struct { + T string `json:"t"` //消息类型 + D interface{} `json:"d"` //传递的消息 } type RenderImageReqMsg struct { - ProductId int64 `json:"product_id"` - SizeId int64 `json:"size_id"` - TemplateId int64 `json:"template_id"` + ProductIds []int64 `json:"product_ids"` //产品 id + TemplateTagId int64 `json:"template_tag_id"` //模板标签id + LogoId int64 `json:"logo_id"` //logoid + AlgorithmVersion string `json:"algorithm_version,optional"` //算法版本 } type RenderImageRspMsg struct { - ProductId int64 `json:"product_id"` - SizeId int64 `json:"size_id"` - TemplateId int64 `json:"template_id"` - Source string `json:"source"` + ProductId int64 `json:"product_id"` //产品 id + TemplateTagId int64 `json:"template_tag_id"` //模板标签id + AlgorithmVersion string `json:"algorithm_version,optional"` //算法版本 + LogoId int64 `json:"logo_id"` //logoid + Image string `json:"image"` //渲染后的图片 +} + +type ThirdPartyLoginRspMsg struct { + Token string `json:"token"` } type RenderNotifyReq struct { - Sign string `json:"sign"` - Time int64 `json:"time"` - NotifyList []NotifyItem `json:"notify_list"` + Sign string `json:"sign"` + Time int64 `json:"time"` + Info NotifyInfo `json:"info"` } -type NotifyItem struct { - ProductId int64 `json:"product_id"` - SizeId int64 `json:"size_id"` - TemplateId int64 `json:"template_id"` - Source string `json:"source"` +type NotifyInfo struct { + ProductId int64 `json:"product_id"` //产品id + TemplateTagId int64 `json:"template_tag_id"` //模板标签id + AlgorithmVersion string `json:"algorithm_version,optional"` //算法版本 + LogoId int64 `json:"logo_id"` //logoid + Image string `json:"image"` +} + +type ThirdPartyLoginNotifyReq struct { + Sign string `json:"sign"` + Time int64 `json:"time"` + Info ThirdPartyLoginNotify `json:"info"` +} + +type ThirdPartyLoginNotify struct { + WebsocketId uint64 `json:"websocket_id"` + Token string `json:"token"` } type Request struct { diff --git a/server_api/product-template-tag.api b/server_api/product-template-tag.api index ae62099a..c8b1a2cc 100644 --- a/server_api/product-template-tag.api +++ b/server_api/product-template-tag.api @@ -12,7 +12,7 @@ import "basic.api" service product-template-tag { //获取产品模板标签列表 @handler GetProductTemplateTagsHandler - get /api/product-template/get_product_template_tags(GetProductTemplateTagsReq) returns (response); + get /api/product-template-tag/get_product_template_tags(GetProductTemplateTagsReq) returns (response); } //获取产品模板标签列表 @@ -20,6 +20,7 @@ type GetProductTemplateTagsReq { Limit int `form:"limit"` } type GetProductTemplateTagsRsp { + Id int64 `json:"id"` Tag string `json:"tag"` Cover string `json:"cover"` } \ No newline at end of file diff --git a/server_api/websocket.api b/server_api/websocket.api index d0daf98b..e24e2874 100644 --- a/server_api/websocket.api +++ b/server_api/websocket.api @@ -11,41 +11,56 @@ import "basic.api" service websocket { //websocket数据交互 @handler DataTransferHandler - get /api/websocket/data_transfer(DataTransferReq) returns (response); - //渲染完了通知接口 + get /api/websocket/data_transfer(request) returns (response); + //云渲染完了通知接口 @handler RenderNotifyHandler post /api/websocket/render_notify(RenderNotifyReq) returns (response); + //第三方登录通知接口 + @handler ThirdPartyLoginNotifyHandler + post /api/websocket/third_party_login_notify(ThirdPartyLoginNotifyReq) returns (response); } //websocket数据交互 -type DataTransferReq { - T string `json:"t"` //消息类型 - D string `json:"d"` //传递的消息 +type DataTransferData { + T string `json:"t"` //消息类型 + D interface{} `json:"d"` //传递的消息 } -type DataTransferRsp { - T string `json:"t"` //消息类型 - D string `json:"d"` //传递的消息 -} -type RenderImageReqMsg { //websocket接受需要云渲染的图片 - ProductId int64 `json:"product_id"` - SizeId int64 `json:"size_id"` - TemplateId int64 `json:"template_id"` +type RenderImageReqMsg { //websocket接受要云渲染处理的数据 + ProductIds []int64 `json:"product_ids"` //产品 id + TemplateTagId int64 `json:"template_tag_id"` //模板标签id + LogoId int64 `json:"logo_id"` //logoid + AlgorithmVersion string `json:"algorithm_version,optional"` //算法版本 } type RenderImageRspMsg { //websocket发送渲染完的数据 - ProductId int64 `json:"product_id"` - SizeId int64 `json:"size_id"` - TemplateId int64 `json:"template_id"` - Source string `json:"source"` + ProductId int64 `json:"product_id"` //产品 id + TemplateTagId int64 `json:"template_tag_id"` //模板标签id + AlgorithmVersion string `json:"algorithm_version,optional"` //算法版本 + LogoId int64 `json:"logo_id"` //logoid + Image string `json:"image"` //渲染后的图片 +} +type ThirdPartyLoginRspMsg { //websocket三方登录的通知数据 + Token string `json:"token"` } //渲染完了通知接口 type RenderNotifyReq { - Sign string `json:"sign"` - Time int64 `json:"time"` - NotifyList []NotifyItem `json:"notify_list"` + Sign string `json:"sign"` + Time int64 `json:"time"` + Info NotifyInfo `json:"info"` } -type NotifyItem { - ProductId int64 `json:"product_id"` - SizeId int64 `json:"size_id"` - TemplateId int64 `json:"template_id"` - Source string `json:"source"` +type NotifyInfo { + ProductId int64 `json:"product_id"` //产品id + TemplateTagId int64 `json:"template_tag_id"` //模板标签id + AlgorithmVersion string `json:"algorithm_version,optional"` //算法版本 + LogoId int64 `json:"logo_id"` //logoid + Image string `json:"image"` +} +//第三方登录通知接口 +type ThirdPartyLoginNotifyReq { + Sign string `json:"sign"` + Time int64 `json:"time"` + Info ThirdPartyLoginNotify `json:"info"` +} +type ThirdPartyLoginNotify { + WebsocketId uint64 `json:"websocket_id"` + Token string `json:"token"` } \ No newline at end of file diff --git a/utils/id_generator/wesocket.go b/utils/id_generator/wesocket.go new file mode 100644 index 00000000..62f21adf --- /dev/null +++ b/utils/id_generator/wesocket.go @@ -0,0 +1,23 @@ +package id_generator + +import "sync" + +type WebsocketId struct { + nodeId uint64 + count uint64 + mu sync.Mutex +} + +func (wid *WebsocketId) Get() uint64 { + wid.mu.Lock() + defer wid.mu.Unlock() + wid.count++ + return (wid.count << 8) | wid.nodeId +} + +func NewWebsocketId(NodeId uint8) *WebsocketId { + return &WebsocketId{ + nodeId: uint64(NodeId), + count: 0, + } +}