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

This commit is contained in:
momo 2023-08-23 11:09:26 +08:00
commit 9460504488
2 changed files with 39 additions and 14 deletions

View File

@ -146,6 +146,11 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.User
//保存连接 //保存连接
mapConnPool.Store(uniqueId, ws) mapConnPool.Store(uniqueId, ws)
go func() { go func() {
defer func() {
if err := recover(); err != nil {
logx.Error("set conn pool panic:", err)
}
}()
//把连接成功消息发回去 //把连接成功消息发回去
time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究) time.Sleep(time.Second * 1) //兼容下火狐(直接发回去收不到第一条消息:有待研究)
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, uniqueId)) ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, uniqueId))
@ -205,7 +210,7 @@ func (l *DataTransferLogic) unAuthResponse(conn *websocket.Conn) {
conn.Close() conn.Close()
} }
// 心跳 // 心跳检测
func (w *wsConnectItem) heartbeat() { func (w *wsConnectItem) heartbeat() {
tick := time.Tick(time.Second * 5) tick := time.Tick(time.Second * 5)
for { for {
@ -223,7 +228,7 @@ func (w *wsConnectItem) heartbeat() {
} }
} }
// 关闭连接 // 关闭websocket连接
func (w *wsConnectItem) close() { func (w *wsConnectItem) close() {
w.mutex.Lock() w.mutex.Lock()
defer w.mutex.Unlock() defer w.mutex.Unlock()
@ -239,8 +244,13 @@ func (w *wsConnectItem) close() {
logx.Info("websocket:", w.uniqueId, " is closed") logx.Info("websocket:", w.uniqueId, " is closed")
} }
// 读取输出返回给客户 // 读取出口缓冲池数据输出返回给浏览器
func (w *wsConnectItem) writeLoop() { func (w *wsConnectItem) writeLoop() {
defer func() {
if err := recover(); err != nil {
logx.Error("write loop panic:", err)
}
}()
for { for {
select { select {
case <-w.closeChan: //如果关闭了 case <-w.closeChan: //如果关闭了
@ -255,8 +265,13 @@ func (w *wsConnectItem) writeLoop() {
} }
} }
// 接受客户端发来的消息 // 接受客户端发来的消息并写入入口缓冲池
func (w *wsConnectItem) readLoop() { func (w *wsConnectItem) readLoop() {
defer func() {
if err := recover(); err != nil {
logx.Error("read loop panic:", err)
}
}()
for { for {
select { select {
case <-w.closeChan: //如果关闭了 case <-w.closeChan: //如果关闭了
@ -278,8 +293,13 @@ func (w *wsConnectItem) readLoop() {
} }
} }
// 把收到的消息发往不同的地方处理 // 消费websocket入口数据池中的数据
func (w *wsConnectItem) sendLoop() { func (w *wsConnectItem) sendLoop() {
defer func() {
if err := recover(); err != nil {
logx.Error("send loop panic:", err)
}
}()
for { for {
select { select {
case <-w.closeChan: case <-w.closeChan:
@ -290,7 +310,7 @@ func (w *wsConnectItem) sendLoop() {
} }
} }
// 把要传递给客户端的数据放入outchan // 把要传递给客户端的数据放入出口缓冲池
func (w *wsConnectItem) sendToOutChan(data []byte) { func (w *wsConnectItem) sendToOutChan(data []byte) {
select { select {
case <-w.closeChan: case <-w.closeChan:
@ -302,7 +322,7 @@ func (w *wsConnectItem) sendToOutChan(data []byte) {
} }
} }
// 格式化返回数据 // 格式化为websocket标准返回格式
func (w *wsConnectItem) respondDataFormat(msgType string, data interface{}) []byte { func (w *wsConnectItem) respondDataFormat(msgType string, data interface{}) []byte {
d := websocket_data.DataTransferData{ d := websocket_data.DataTransferData{
T: msgType, T: msgType,
@ -312,7 +332,7 @@ func (w *wsConnectItem) respondDataFormat(msgType string, data interface{}) []by
return b return b
} }
// 处理接受到的数据 // 处理入口缓冲池中不同类型的数据(分发处理)
func (w *wsConnectItem) dealwithReciveData(data []byte) { func (w *wsConnectItem) dealwithReciveData(data []byte) {
var parseInfo websocket_data.DataTransferData var parseInfo websocket_data.DataTransferData
if err := json.Unmarshal(data, &parseInfo); err != nil { if err := json.Unmarshal(data, &parseInfo); err != nil {

View File

@ -133,6 +133,7 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
resource, err := w.logic.svcCtx.AllModels.FsResource.FindOneById(w.logic.ctx, taskId) resource, err := w.logic.svcCtx.AllModels.FsResource.FindOneById(w.logic.ctx, taskId)
if err != nil { if err != nil {
if !errors.Is(err, gorm.ErrRecordNotFound) { if !errors.Is(err, gorm.ErrRecordNotFound) {
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE_ERR, fmt.Sprintf("获取云渲染缓存资源错误 task_id:%s ", taskId)))
logx.Error("failed to find render resource:", err) logx.Error("failed to find render resource:", err)
return return
} }
@ -164,11 +165,6 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
// 组装数据发送给unity // 组装数据发送给unity
func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.RenderImageReqMsg) error { func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.RenderImageReqMsg) error {
defer func() {
if err := recover(); err != nil {
logx.Error("assembleRenderData panic:", err)
}
}()
//获取产品第一个尺寸 //获取产品第一个尺寸
productFirstSize, err := w.logic.svcCtx.AllModels.FsProductSize.GetProductFirstSize(w.logic.ctx, info.RenderData.ProductId) productFirstSize, err := w.logic.svcCtx.AllModels.FsProductSize.GetProductFirstSize(w.logic.ctx, info.RenderData.ProductId)
if err != nil { if err != nil {
@ -389,6 +385,10 @@ func (w *wsConnectItem) deleteRenderTask(data renderImageControlChanItem) {
// 修改耗时属性(只有耗时属性可以更新) // 修改耗时属性(只有耗时属性可以更新)
func (w *wsConnectItem) modifyRenderTaskTimeConsuming(data renderImageControlChanItem) { func (w *wsConnectItem) modifyRenderTaskTimeConsuming(data renderImageControlChanItem) {
if data.TaskId == "" {
logx.Error("设置耗时属性需要的task_id不能为空")
return
}
//强制设为修改耗时属性 //强制设为修改耗时属性
data.Option = 2 data.Option = 2
select { select {
@ -401,8 +401,13 @@ func (w *wsConnectItem) modifyRenderTaskTimeConsuming(data renderImageControlCha
} }
} }
// 操作连接中渲染任务的增加/删除任务map不能读写并发所以放在chan里面串行执行 // 处理渲染任务的增加/删除/修改耗时属性任务map不能读写并发所以放在chan里面串行执行
func (w *wsConnectItem) operationRenderTask() { func (w *wsConnectItem) operationRenderTask() {
defer func() {
if err := recover(); err != nil {
logx.Error("operation render task panic:", err)
}
}()
for { for {
select { select {
case <-w.closeChan: case <-w.closeChan: