This commit is contained in:
laodaming 2023-08-07 19:13:16 +08:00
parent 84ac7d92b2
commit 31a3966e39
4 changed files with 29 additions and 9 deletions

View File

@ -1,6 +1,8 @@
package consumer
import (
"encoding/json"
"fusenapi/utils/websocket_data"
"github.com/zeromicro/go-zero/core/logx"
)
@ -10,5 +12,11 @@ type MqConsumerRenderAssemble struct {
func (m *MqConsumerRenderAssemble) Run(data []byte) error {
logx.Info("收到需要组装的消息:", string(data))
var parseInfo websocket_data.AssembleRenderData
if err := json.Unmarshal(data, &parseInfo); err != nil {
logx.Error("MqConsumerRenderAssemble数据格式错误:", err)
return nil //不返回错误就删除消息
}
return nil
}

View File

@ -65,12 +65,13 @@ var (
type wsConnectItem struct {
conn *websocket.Conn //websocket的连接
rabbitMq *initalize.RabbitMqHandle
closeChan chan struct{} //ws连接关闭chan
isClose bool //是否已经关闭
uniqueId uint64 //ws连接唯一标识
inChan chan []byte //接受消息缓冲通道
outChan chan []byte //发送回客户端的消息
mutex sync.Mutex //互斥锁
closeChan chan struct{} //ws连接关闭chan
isClose bool //是否已经关闭
uniqueId uint64 //ws连接唯一标识
inChan chan []byte //接受消息缓冲通道
outChan chan []byte //发送回客户端的消息
mutex sync.Mutex //互斥锁
userId int64
renderProperty renderProperty //扩展云渲染属性
}
@ -83,10 +84,14 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
}
defer conn.Close()
//鉴权不成功10秒后断开
/*isAuth, _ := l.checkAuth(svcCtx, r)
var (
userInfo *auth.UserInfo
isAuth bool
)
isAuth, userInfo = l.checkAuth(svcCtx, r)
if !isAuth {
time.Sleep(time.Second) //兼容下火狐
rsp := types.DataTransferData{
rsp := websocket_data.DataTransferData{
T: constants.WEBSOCKET_UNAUTH,
D: nil,
}
@ -96,7 +101,7 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
//发送关闭信息
_ = conn.WriteMessage(websocket.CloseMessage, nil)
return
}*/
}
//生成连接唯一标识
uniqueId := websocketIdGenerator.Get()
ws := wsConnectItem{
@ -111,6 +116,11 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
renderImageTaskCtlChan: make(chan renderImageControlChanItem, 100),
},
}
if userInfo.UserId > 0 {
ws.userId = userInfo.UserId
} else {
ws.userId = userInfo.GuestId
}
//保存连接
mapConnPool.Store(uniqueId, ws)
defer ws.close()

View File

@ -39,6 +39,7 @@ func (w *wsConnectItem) renderImage(data []byte) {
}
tmpData := websocket_data.AssembleRenderData{
TaskId: taskId,
UserId: w.userId,
RenderData: renderImageData.RenderData,
}
d, _ := json.Marshal(tmpData)

View File

@ -37,5 +37,6 @@ type ThirdPartyLoginRspMsg struct {
// 发送到渲染组装的mq数据
type AssembleRenderData struct {
TaskId string `json:"task_id"`
UserId int64 `json:"user_id"`
RenderData RenderData `json:"render_data"`
}