fusenapi/server/websocket/internal/logic/ws_render_image_logic.go

package logic

import (
	"encoding/json"
	"errors"
	"fusenapi/constants"
	"fusenapi/utils/hash"
	"fusenapi/utils/websocket_data"
	"github.com/zeromicro/go-zero/core/logx"
	"gorm.io/gorm"
)

// 云渲染属性
type renderProperty struct {
	renderImageTask        map[string]string               //需要渲染的图片任务 key是taskId val 是renderId
	renderImageTaskCtlChan chan renderImageControlChanItem //渲染任务新增移除的控制通道
}

// 渲染任务新增移除的控制通道的数据
type renderImageControlChanItem struct {
	Option   int    // 0删除 1添加
	TaskId   string //map的key
	RenderId string // map的val
}

// 渲染发送到组装数据组装数据
func (w *wsConnectItem) renderImage(data []byte) {
	var renderImageData websocket_data.AssembleRenderData
	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)
	//用户id赋值
	renderImageData.RenderData.UserId = w.userId
	renderImageData.RenderData.GuestId = w.guestId
	//生成任务id
	taskId := hash.JsonHashKey(renderImageData.RenderData)
	//查询有没有缓存的资源,有就返回######################
	resource, err := w.allModels.FsResource.FindOneById(w.ctx, taskId)
	if err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			logx.Error("failed to find render resource:", err)
			return
		}
	} else {
		//返回给客户端
		b := w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
			RenderId: renderImageData.RenderId,
			Image:    *resource.ResourceUrl,
		})
		//发送数据到out chan
		w.sendToOutChan(b)
		return
	}
	//###########################################
	//把需要渲染的图片任务加进去
	w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
		Option:   1, //0删除 1添加
		TaskId:   taskId,
		RenderId: renderImageData.RenderId,
	}
	tmpData := websocket_data.AssembleRenderData{
		TaskId:     taskId,
		RenderId:   renderImageData.RenderId,
		RenderData: renderImageData.RenderData,
	}
	d, _ := json.Marshal(tmpData)
	//发送给对应的流水线组装数据
	if err := w.rabbitMq.SendMsg(constants.RABBIT_MQ_ASSEMBLE_RENDER_DATA, d); err != nil {
		logx.Error("发送渲染任务数据到MQ失败:", string(data), "err:", err)
		return
	}
	logx.Info("发送渲染数据到rabbitmq成功:", string(data))
}

// 操作连接中渲染任务的增加/删除
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.TaskId)
			case 1: //新增任务
				w.renderProperty.renderImageTask[data.TaskId] = data.RenderId
			}
		}
	}
}