Merge branch 'develop' into feature/auth

This commit is contained in:
eson
2023-08-22 10:34:21 +08:00
79 changed files with 1360 additions and 495 deletions

View File

@@ -3,7 +3,6 @@ package logic
import (
"bytes"
"encoding/json"
"fmt"
"fusenapi/constants"
"fusenapi/utils/auth"
"fusenapi/utils/id_generator"
@@ -81,7 +80,13 @@ type wsConnectItem struct {
renderProperty renderProperty //扩展云渲染属性
}
// 请求建立连接升级websocket协议
func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request) {
//把子协议携带的token设置到标准token头信息中
token := r.Header.Get("Sec-Websocket-Protocol")
r.Header.Set("Authorization", "Bearer "+token)
//设置Sec-Websocket-Protocol
upgrade.Subprotocols = []string{token}
//升级websocket
conn, err := upgrade.Upgrade(w, r, nil)
if err != nil {
@@ -90,29 +95,18 @@ func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request)
}
defer conn.Close()
//鉴权不成功后断开
/*var (
var (
userInfo *auth.UserInfo
isAuth bool
)
isAuth, userInfo = l.checkAuth(r)
if !isAuth {
time.Sleep(time.Second * 1) //兼容下火狐
rsp := websocket_data.DataTransferData{
T: constants.WEBSOCKET_UNAUTH,
D: nil,
}
b, _ := json.Marshal(rsp)
//先发一条正常信息
_ = conn.WriteMessage(websocket.TextMessage, b)
//发送关闭信息
_ = conn.WriteMessage(websocket.CloseMessage, nil)
//未授权响应消息
l.unAuthResponse(conn)
return
}*/
//测试的目前写死 39
var userInfo auth.UserInfo
userInfo.UserId = 39
}
//设置连接
ws := l.setConnPool(conn, userInfo)
ws := l.setConnPool(conn, *userInfo)
defer ws.close()
//循环读客户端信息
go ws.readLoop()
@@ -144,7 +138,7 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.User
userId: userInfo.UserId,
guestId: userInfo.GuestId,
renderProperty: renderProperty{
renderImageTask: make(map[string]string),
renderImageTask: make(map[string]*renderTask),
renderImageTaskCtlChan: make(chan renderImageControlChanItem, 100),
renderChan: make(chan []byte, 100),
},
@@ -153,7 +147,7 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.User
mapConnPool.Store(uniqueId, ws)
go func() {
//把连接成功消息发回去
time.Sleep(time.Second * 1) //兼容下火狐
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究)
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, uniqueId))
}()
return ws
@@ -162,24 +156,20 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.User
// 获取唯一id
func (l *DataTransferLogic) getUniqueId(userInfo auth.UserInfo) string {
//后面拼接上用户id
uniqueId := uuid.New().String() + getUserPart(userInfo.UserId, userInfo.GuestId)
uniqueId := uuid.New().String() + getUserJoinPart(userInfo.UserId, userInfo.GuestId)
if _, ok := mapConnPool.Load(uniqueId); ok {
uniqueId = l.getUniqueId(userInfo)
}
return uniqueId
}
// 获取用户拼接部分
func getUserPart(userId, guestId int64) string {
return fmt.Sprintf("_%d_%d", userId, guestId)
}
// 鉴权
func (l *DataTransferLogic) checkAuth(r *http.Request) (isAuth bool, userInfo *auth.UserInfo) {
// 解析JWT token,并对空用户进行判断
claims, err := l.svcCtx.ParseJwtToken(r)
// 如果解析JWT token出错,则返回未授权的JSON响应并记录错误消息
if err != nil {
logx.Error(err)
return false, nil
}
if claims != nil {
@@ -187,6 +177,7 @@ func (l *DataTransferLogic) checkAuth(r *http.Request) (isAuth bool, userInfo *a
userInfo, err = auth.GetUserInfoFormMapClaims(claims)
// 如果获取用户信息出错,则返回未授权的JSON响应并记录错误消息
if err != nil {
logx.Error(err)
return false, nil
}
//不是登录用户也不是游客
@@ -198,6 +189,22 @@ func (l *DataTransferLogic) checkAuth(r *http.Request) (isAuth bool, userInfo *a
return false, nil
}
// 鉴权失败通知
func (l *DataTransferLogic) unAuthResponse(conn *websocket.Conn) {
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究)
rsp := websocket_data.DataTransferData{
T: constants.WEBSOCKET_UNAUTH,
D: nil,
}
b, _ := json.Marshal(rsp)
//先发一条正常信息
_ = conn.WriteMessage(websocket.TextMessage, b)
//发送关闭信息
_ = conn.WriteMessage(websocket.CloseMessage, nil)
//关闭连接
conn.Close()
}
// 心跳
func (w *wsConnectItem) heartbeat() {
tick := time.Tick(time.Second * 5)

View File

@@ -1,18 +1,14 @@
package logic
import (
"fusenapi/constants"
"context"
"fusenapi/server/websocket/internal/svc"
"fusenapi/server/websocket/internal/types"
"fusenapi/utils/auth"
"fusenapi/utils/basic"
"fusenapi/utils/file"
"fusenapi/utils/websocket_data"
"time"
"context"
"fusenapi/server/websocket/internal/svc"
"fusenapi/server/websocket/internal/types"
"github.com/zeromicro/go-zero/core/logx"
)
@@ -41,14 +37,17 @@ func NewRenderNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Rend
func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) {
if req.TaskId == "" {
logx.Error("渲染回调参数错误invalid param task_id")
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param task_id")
}
if req.Image == "" {
logx.Error("渲染回调参数错误invalid param image")
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param image")
}
if req.UserId == 0 && req.GuestId == 0 {
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid user_id or guest_id")
}
//存base64打印测试
/* f, _ := os.Create("b.txt")
defer f.Close()
f.WriteString(req.Image)*/
// 上传文件
var upload = file.Upload{
Ctx: l.ctx,
@@ -67,7 +66,7 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *a
FileByte: nil,
})
if err != nil {
logx.Error(err)
logx.Error("渲染回调上传文件失败:", err)
return resp.SetStatusWithMessage(basic.CodeFileUploadErr, "failed to upload render resource image")
}
//遍历websocket链接把数据传进去
@@ -75,37 +74,24 @@ func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *a
//断言连接
ws, ok := value.(wsConnectItem)
if !ok {
logx.Error("渲染回调断言websocket连接失败")
return true
}
//关闭标识
if ws.isClose {
return true
}
//查询有无该渲染任务
renderId, ok := ws.renderProperty.renderImageTask[req.TaskId]
if !ok {
return true
}
b := ws.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
RenderId: renderId,
Image: uploadRes.ResourceUrl,
//记录收到unity渲染结果时间
ws.modifyRenderTaskTimeConsuming(renderImageControlChanItem{
Option: 2,
TaskProperty: renderTask{
UnityRenderEndTime: time.Now().UTC().Unix(),
},
})
//发送处理并删除任务
ws.deleteRenderTask(renderImageControlChanItem{
Option: 0, //0删除 1添加
TaskId: req.TaskId,
RenderNotifyImageUrl: uploadRes.ResourceUrl,
})
deleteTask := renderImageControlChanItem{
Option: 0, //0删除 1添加
TaskId: req.TaskId,
RenderId: renderId,
}
select {
case <-ws.closeChan: //关闭了
return true
case ws.renderProperty.renderImageTaskCtlChan <- deleteTask: //删除对应的需要渲染的图片map
//发送数据到out chan
ws.sendToOutChan(b)
case <-time.After(time.Second * 3): //超时丢弃
return true
}
return true
})
logx.Info("渲染回调成功######################")
logx.Info("渲染回调成功,渲染结果图片为:", uploadRes.ResourceUrl)
return resp.SetStatusWithMessage(basic.CodeOK, "success")
}

View File

@@ -4,6 +4,7 @@ import (
"bytes"
"encoding/json"
"errors"
"fmt"
"fusenapi/constants"
"fusenapi/service/repositories"
"fusenapi/utils/curl"
@@ -18,16 +19,25 @@ import (
// 云渲染属性
type renderProperty struct {
renderImageTask map[string]string //需要渲染的图片任务 key是taskId val 是renderId
renderImageTask map[string]*renderTask //需要渲染的图片任务 key是taskId val 是renderId
renderImageTaskCtlChan chan renderImageControlChanItem //渲染任务新增移除的控制通道
renderChan chan []byte //渲染的缓冲队列
}
type renderTask struct {
RenderId string //渲染id(前端传的)
CombineBeginTime int64 //合图开始时间
CombineEndTime int64 //合图结束时间
UnityRenderBeginTime int64 //发送给unity时间
UnityRenderEndTime int64 //unity回调结果时间
}
// 渲染任务新增移除的控制通道的数据
type renderImageControlChanItem struct {
Option int // 0删除 1添加
TaskId string //map的key
RenderId string // map的val
Option int // 0删除 1添加
TaskId string //map的key
RenderId string // map的val(增加任务时候传)
RenderNotifyImageUrl string //渲染回调数据(删除任务时候传)
TaskProperty renderTask //渲染任务的属性
}
// 发送到渲染缓冲池
@@ -35,14 +45,14 @@ func (w *wsConnectItem) sendToRenderChan(data []byte) {
select {
case <-w.closeChan: //已经关闭
return
case w.renderProperty.renderChan <- data:
case w.renderProperty.renderChan <- data: //发入到缓冲池
return
case <-time.After(time.Second * 3):
case <-time.After(time.Second * 3): //三秒没进入缓冲池就丢弃
return
}
}
// 渲染发送到组装数据组装数据
// 渲染发送到组装数据组装数据(缓冲池)
func (w *wsConnectItem) renderImage() {
defer func() {
if err := recover(); err != nil {
@@ -69,7 +79,6 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
logx.Error("invalid format of websocket render image message", err)
return
}
logx.Info("收到请求云渲染图片数据:", renderImageData)
if renderImageData.RenderId == "" {
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_ERR_DATA_FORMAT, "invalid format of websocket render image message:render_id is empty"))
logx.Error("invalid format of websocket render image message:render_id is empty")
@@ -116,7 +125,10 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
hashVal := renderImageData.RenderData
hashVal.UserId = 0
hashVal.GuestId = 0
taskId := hash.JsonHashKey(hashVal)
hashByte, _ := json.Marshal(hashVal)
var hashData map[string]interface{}
_ = json.Unmarshal(hashByte, &hashData)
taskId := hash.JsonHashKey(hashData)
//查询有没有缓存的资源,有就返回######################
resource, err := w.logic.svcCtx.AllModels.FsResource.FindOneById(w.logic.ctx, taskId)
if err != nil {
@@ -127,8 +139,10 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
} else {
//返回给客户端
b := w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
RenderId: renderImageData.RenderId,
Image: *resource.ResourceUrl,
RenderId: renderImageData.RenderId,
Image: *resource.ResourceUrl,
CombineTakesTime: "耗时0秒(缓存)",
UnityRenderTakesTime: "耗时0秒(缓存)",
})
//发送数据到out chan
w.sendToOutChan(b)
@@ -136,11 +150,11 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
}
//###########################################
//把需要渲染的图片任务加进去
w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
w.createRenderTask(renderImageControlChanItem{
Option: 1, //0删除 1添加
TaskId: taskId,
RenderId: renderImageData.RenderId,
}
})
//组装数据
if err = w.assembleRenderData(taskId, renderImageData); err != nil {
logx.Error("组装数据失败:", err)
@@ -165,8 +179,15 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
logx.Error("failed to get template info:", err)
return err
}
//记录刀版图合成开始时间
w.modifyRenderTaskTimeConsuming(renderImageControlChanItem{
Option: 2,
TaskProperty: renderTask{
CombineBeginTime: time.Now().UTC().Unix(),
},
})
//获取刀版图
res, err := w.logic.svcCtx.Repositories.ImageHandle.LogoCombine(w.logic.ctx, &repositories.LogoCombineReq{
combineReq := repositories.LogoCombineReq{
UserId: info.RenderData.UserId,
GuestId: info.RenderData.GuestId,
TemplateId: productTemplate.Id,
@@ -175,9 +196,10 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
Slogan: info.RenderData.Slogan,
Address: info.RenderData.Address,
Phone: info.RenderData.Phone,
})
}
res, err := w.logic.svcCtx.Repositories.ImageHandle.LogoCombine(w.logic.ctx, &combineReq)
if err != nil {
logx.Error("合成刀版图失败:", err)
logx.Error("合成刀版图失败,合成请求数据:", combineReq, "错误信息:", err)
return err
}
combineImage := "" //刀版图
@@ -187,17 +209,18 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
logx.Error("合成刀版图失败,合成的刀版图是空指针:", err)
return err
}
logx.Info("合成刀版图成功:", *res.ResourceUrl)
//记录刀版图合成结束时间
w.modifyRenderTaskTimeConsuming(renderImageControlChanItem{
Option: 2,
TaskProperty: renderTask{
CombineEndTime: time.Now().UTC().Unix(),
},
})
logx.Info("合成刀版图成功,合成刀版图数据:", combineReq, ",logo图片:", info.RenderData.Logo, " 刀版图:", *res.ResourceUrl)
//获取渲染设置信息
element, err := w.logic.svcCtx.AllModels.FsProductTemplateElement.FindOneByModelId(w.logic.ctx, *productTemplate.ModelId)
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
// todo 没有图就给他返回一张默认(后面要删除)
defaultImg := w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
RenderId: info.RenderId,
Image: "https://s3.us-west-1.amazonaws.com/storage.fusenpack.com/695463af6e9b93c003db39ddf728241f9523efc55b20dc37f30fe5d96ed54fb5",
})
w.sendToOutChan(defaultImg)
logx.Error("element info is not found,model_id = ", *productTemplate.ModelId)
return err
}
@@ -208,6 +231,9 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
refletion := -1
if element.Refletion != nil && *element.Refletion != "" {
refletion, err = strconv.Atoi(*element.Refletion)
if err != nil {
logx.Error("err refletion:set default -1")
}
}
//组装data数据
var mode map[string]interface{}
@@ -272,7 +298,7 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
"folder": "", //todo 千人千面需要使用
}
//请求unity接口
url := "http://api.fusen.3718.cn:4050/api/render/queue/push"
url := w.logic.svcCtx.Config.Unity.Host + "/api/render/queue/push"
header := make(map[string]string)
header["content-type"] = "application/json"
t := time.Now().UTC()
@@ -283,17 +309,66 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
"create_at": t,
"render_data": sendData,
}
p, _ := json.Marshal(postData)
_, err = curl.ApiCall(url, "POST", header, bytes.NewReader(p), time.Second*10)
postDataBytes, _ := json.Marshal(postData)
_, err = curl.ApiCall(url, "POST", header, bytes.NewReader(postDataBytes), time.Second*10)
if err != nil {
logx.Error("failed to send data to unity")
return err
}
logx.Info("发送到unity成功################")
//记录发送到unity时间
w.modifyRenderTaskTimeConsuming(renderImageControlChanItem{
Option: 2,
TaskProperty: renderTask{
UnityRenderBeginTime: time.Now().UTC().Unix(),
},
})
logx.Info("发送到unity成功,刀版图:", combineImage, " 请求unity的数据:", string(postDataBytes))
return nil
}
// 操作连接中渲染任务的增加/删除
// 增加渲染任务
func (w *wsConnectItem) createRenderTask(data renderImageControlChanItem) {
//强制设为增加
data.Option = 1
select {
case <-w.closeChan: //关闭
return
case w.renderProperty.renderImageTaskCtlChan <- data:
return
case <-time.After(time.Second * 3):
return
}
}
// 渲染回调处理并删除渲染任务
func (w *wsConnectItem) deleteRenderTask(data renderImageControlChanItem) {
//强制设为删除
data.Option = 0
select {
case <-w.closeChan: //关闭
return
case w.renderProperty.renderImageTaskCtlChan <- data:
return
case <-time.After(time.Second * 3):
return
}
}
// 修改耗时属性(只有耗时属性可以更新)
func (w *wsConnectItem) modifyRenderTaskTimeConsuming(data renderImageControlChanItem) {
//强制设为修改耗时属性
data.Option = 2
select {
case <-w.closeChan: //关闭
return
case w.renderProperty.renderImageTaskCtlChan <- data:
return
case <-time.After(time.Second * 3):
return
}
}
// 操作连接中渲染任务的增加/删除任务map不能读写并发所以放在chan里面串行执行
func (w *wsConnectItem) operationRenderTask() {
for {
select {
@@ -301,10 +376,46 @@ func (w *wsConnectItem) operationRenderTask() {
return
case data := <-w.renderProperty.renderImageTaskCtlChan:
switch data.Option {
case 0: //删除任务
case 0: //渲染结果回调,删除任务
//存在任务,则发送渲染结果给前端
if taskData, ok := w.renderProperty.renderImageTask[data.TaskId]; ok {
CombineTakesTime := ""
UnityRenderTakesTime := ""
if taskData.CombineBeginTime > 0 && taskData.CombineEndTime > 0 {
CombineTakesTime = fmt.Sprintf("耗时%d秒", taskData.CombineEndTime-taskData.CombineBeginTime)
}
if taskData.UnityRenderBeginTime > 0 && taskData.UnityRenderEndTime > 0 {
UnityRenderTakesTime = fmt.Sprintf("耗时%d秒", taskData.UnityRenderEndTime-taskData.UnityRenderBeginTime)
}
//发送到出口
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
RenderId: taskData.RenderId,
Image: data.RenderNotifyImageUrl,
CombineTakesTime: CombineTakesTime,
UnityRenderTakesTime: UnityRenderTakesTime,
}))
}
delete(w.renderProperty.renderImageTask, data.TaskId)
case 1: //新增任务
w.renderProperty.renderImageTask[data.TaskId] = data.RenderId
w.renderProperty.renderImageTask[data.TaskId] = &renderTask{
RenderId: data.RenderId,
}
case 2: //修改(耗时)属性
if taskData, ok := w.renderProperty.renderImageTask[data.TaskId]; ok {
if data.TaskProperty.CombineBeginTime != 0 {
taskData.CombineBeginTime = data.TaskProperty.CombineBeginTime
}
if data.TaskProperty.CombineEndTime != 0 {
taskData.CombineEndTime = data.TaskProperty.CombineEndTime
}
if data.TaskProperty.UnityRenderBeginTime != 0 {
taskData.UnityRenderBeginTime = data.TaskProperty.UnityRenderBeginTime
}
if data.TaskProperty.UnityRenderEndTime != 0 {
taskData.UnityRenderEndTime = data.TaskProperty.UnityRenderEndTime
}
logx.Info("**********:", taskData)
}
}
}
}

View File

@@ -2,6 +2,7 @@ package logic
import (
"encoding/json"
"fmt"
"fusenapi/constants"
"github.com/zeromicro/go-zero/core/logx"
)
@@ -22,7 +23,7 @@ func (w *wsConnectItem) reuseLastConnect(data []byte) {
return
}
//合成client后缀,不是同个后缀的不能复用
userPart := getUserPart(w.userId, w.guestId)
userPart := getUserJoinPart(w.userId, w.guestId)
lenUserPart := len(userPart)
if lenClientId <= lenUserPart {
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_REQUEST_RESUME_LAST_CONNECT_ERR, "length of client id is to short"))
@@ -36,10 +37,23 @@ func (w *wsConnectItem) reuseLastConnect(data []byte) {
publicMutex.Lock()
defer publicMutex.Unlock()
//存在是不能给他申请重新绑定
if _, ok := mapConnPool.Load(clientId); ok {
rsp := w.respondDataFormat(constants.WEBSOCKET_REQUEST_RESUME_LAST_CONNECT_ERR, "id has bound by other connect ")
w.sendToOutChan(rsp)
return
if v, ok := mapConnPool.Load(clientId); ok {
obj, ok := v.(wsConnectItem)
if !ok {
logx.Error("连接断言失败")
}
//是当前自己占用
if obj.uniqueId == w.uniqueId {
//重新绑定
w.uniqueId = clientId
rsp := w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, clientId)
w.sendToOutChan(rsp)
return
} else {
rsp := w.respondDataFormat(constants.WEBSOCKET_REQUEST_RESUME_LAST_CONNECT_ERR, "id has bound by other connect ")
w.sendToOutChan(rsp)
return
}
}
//重新绑定
w.uniqueId = clientId
@@ -47,3 +61,8 @@ func (w *wsConnectItem) reuseLastConnect(data []byte) {
w.sendToOutChan(rsp)
return
}
// 获取用户拼接部分(复用标识用到)
func getUserJoinPart(userId, guestId int64) string {
return fmt.Sprintf("_%d_%d", userId, guestId)
}