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

Merge branch 'develop' of https://gitee.com/fusenpack/fusenapi into feature/auth

Browse Source
This commit is contained in:
eson
2023-07-28 13:10:32 +08:00
parent 11cc2547c9 ae717f932e
commit 719c41db00
31 changed files with 809 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

@@ -44,14 +44,17 @@ var (
}
//升级websocket
upgrade = websocket.Upgrader{
ReadBufferSize: 1024 * 10, //最大可读取大小 10M
//最大可读取大小 10M
ReadBufferSize: 1024 * 10,
//握手超时时间15s
HandshakeTimeout: time.Second * 15,
//允许跨域
CheckOrigin: func(r *http.Request) bool {
return true
},
WriteBufferPool: &buffPool,
//写的缓存池
WriteBufferPool: &buffPool,
//是否支持压缩
EnableCompression: true,
}
//websocket连接存储
@@ -78,13 +81,14 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
return
}
defer conn.Close()
w.Header().Set("Connection", "Upgrade")
rsp := types.DataTransferData{}
//鉴权不成功10秒后断开
/*isAuth, _ := l.checkAuth(svcCtx, r)
if !isAuth {
time.Sleep(time.Second) //兼容下火狐
rsp.T = constants.WEBSOCKET_UNAUTH
rsp := types.DataTransferData{
T: constants.WEBSOCKET_UNAUTH,
D: nil,
}
b, _ := json.Marshal(rsp)
//先发一条正常信息
_ = conn.WriteMessage(websocket.TextMessage, b)
@@ -98,8 +102,8 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
conn: conn,
uniqueId: uniqueId,
closeChan: make(chan struct{}, 1),
inChan: make(chan []byte, 100),
outChan: make(chan []byte, 100),
inChan: make(chan []byte, 1000),
outChan: make(chan []byte, 1000),
renderProperty: renderProperty{
renderImageTask: make(map[string]struct{}),
renderImageTaskCtlChan: make(chan renderImageControlChanItem, 100),
@@ -110,9 +114,7 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
defer ws.close()
//把连接成功消息发回去
time.Sleep(time.Second) //兼容下火狐
rsp.T = constants.WEBSOCKET_CONNECT_SUCCESS
rsp.D = uniqueId
b, _ := json.Marshal(rsp)
b := ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, uniqueId)
_ = conn.WriteMessage(websocket.TextMessage, b)
//循环读客户端信息
go ws.readLoop()
@@ -243,8 +245,18 @@ func (w *wsConnectItem) sendToOutChan(data []byte) {
}
// 获取需要渲染图片的map key
func (w *wsConnectItem) getRenderImageMapKey(productId, templateTagId int64, algorithmVersion string) string {
return fmt.Sprintf("%d-%d-%s", productId, templateTagId, algorithmVersion)
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
}
// 处理接受到的数据
@@ -252,6 +264,7 @@ 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)

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

@@ -1,10 +1,6 @@
package logic
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"fusenapi/constants"
"fusenapi/utils/basic"
"time"
@@ -45,7 +41,7 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq) (resp *basi
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param time")
}
//验证签名 sha256
notifyByte, _ := json.Marshal(req.Info)
/*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)
@@ -53,7 +49,7 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq) (resp *basi
//fmt.Println(sign)
if req.Sign != sign {
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid sign")
}
}*/
//遍历websocket链接把数据传进去
mapConnPool.Range(func(key, value any) bool {
//断言连接
@@ -61,21 +57,21 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq) (resp *basi
if !ok {
return true
}
renderKey := ws.getRenderImageMapKey(req.Info.ProductId, req.Info.TemplateTagId, req.Info.AlgorithmVersion)
//关闭标识
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
}
rspData := types.DataTransferData{
T: constants.WEBSOCKET_RENDER_IMAGE,
D: types.RenderImageRspMsg{
ProductId: req.Info.ProductId,
TemplateTagId: req.Info.TemplateTagId,
Image: req.Info.Image,
},
}
b, _ := json.Marshal(rspData)
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添加

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

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

@@ -2,6 +2,7 @@ package logic
import (
"encoding/json"
"fusenapi/constants"
"fusenapi/server/websocket/internal/types"
"github.com/zeromicro/go-zero/core/logx"
)
@@ -18,31 +19,6 @@ type renderImageControlChanItem struct {
Key string //map的key
}
// 渲染请求数据处理发送云渲染服务处理
func (w *wsConnectItem) SendToCloudRender(data []byte) {
var renderImageData types.RenderImageReqMsg
if err := json.Unmarshal(data, &renderImageData); err != nil {
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.AlgorithmVersion)
w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
Option: 1, //0删除 1添加
Key: key,
}
// TODO 数据发送给云渲染服务器
}
}
}
// 操作连接中渲染任务的增加/删除
func (w *wsConnectItem) operationRenderTask() {
for {
@@ -61,3 +37,29 @@ func (w *wsConnectItem) operationRenderTask() {
}
}
}
// 渲染请求数据处理发送云渲染服务处理
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 数据发送给云渲染服务器
}
}
}

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

@@ -13,6 +13,7 @@ type DataTransferData struct {
type RenderImageReqMsg struct {
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"` //算法版本
}
@@ -20,9 +21,14 @@ type RenderImageRspMsg 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 ThirdPartyLoginRspMsg struct {
Token string `json:"token"`
}
type RenderNotifyReq struct {
Sign string `json:"sign"`
Time int64 `json:"time"`
@@ -33,9 +39,21 @@ 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 {
}