fusen/fusenapi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

分支合并

Browse Source
This commit is contained in:
Hiven 2023-07-28 11:20:12 +08:00
commit 335ecc1f41
18 changed files with 619 additions and 391 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
constants/websocket.go
View File

@ -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"

1
model/gmodel/fs_pay_gen.go
View File

@ -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"` // 第三方支付编号

6
model/gmodel/var_gen.go
View File

@ -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),

2
server/product-template-tag/internal/handler/routes.go
View File

@ -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),
},
},

1
server/product-template-tag/internal/logic/getproducttemplatetagslogic.go
View File

@ -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,
})

1
server/product-template-tag/internal/types/types.go
View File

@ -10,6 +10,7 @@ type GetProductTemplateTagsReq struct {
}
type GetProductTemplateTagsRsp struct {
Id int64 `json:"id"`
Tag string `json:"tag"`
Cover string `json:"cover"`
}

257
server/websocket/internal/handler/datatransferhandler.go
View File

@ -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
l.DataTransfer(svcCtx, w, r)
}
}
//生成连接唯一标识
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()
}
}
// 心跳
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)
}

74
server/websocket/internal/handler/rendernotifyhandler.go
View File

@ -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
}
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,
})
// 创建一个业务逻辑层实例
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)
}
}
}

5
server/websocket/internal/handler/routes.go
View File

@ -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),
},
},
)
}

35
server/websocket/internal/handler/thirdpartyloginnotifyhandler.go Normal file
View File

@ -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)
}
}
}

270
server/websocket/internal/logic/datatransferlogic.go
View File

@ -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) {
// }
// 处理逻辑后 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)
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{}
)
// 每个连接的连接基本属性
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:
}
}

52
server/websocket/internal/logic/rendernotifylogic.go
View File

@ -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)
}

78
server/websocket/internal/logic/thirdpartyloginnotifylogic.go Normal file
View File

@ -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")
}
}

65
server/websocket/internal/logic/ws_render_image_logic.go Normal file
View File

@ -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 数据发送给云渲染服务器
}
}
}

53
server/websocket/internal/types/types.go
View File

@ -5,40 +5,53 @@ import (
"fusenapi/utils/basic"
)
type DataTransferReq struct {
type DataTransferData struct {
T string `json:"t"` //消息类型
D string `json:"d"` //传递的消息
}
type DataTransferRsp struct {
T string `json:"t"` //消息类型
D string `json:"d"` //传递的消息
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"`
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 {

3
server_api/product-template-tag.api
View File

@ -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"`
}

59
server_api/websocket.api
View File

@ -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 {
type DataTransferData {
T string `json:"t"` //消息类型
D string `json:"d"` //传递的消息
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"`
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"`
}

23
utils/id_generator/wesocket.go Normal file
View File

@ -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,
}
}