fix

constants/websocket.go

@@ -8,6 +8,8 @@ const (
 	WEBSOCKET_UNAUTH = "WEBSOCKET_UNAUTH"
 	//ws连接成功
 	WEBSOCKET_CONNECT_SUCCESS = "WEBSOCKET_CONNECT_SUCCESS"
+	//渲染前数据组装
+	WEBSOCKET_RENDER_IMAGE_ASSEMBLE = "WEBSOCKET_RENDER_IMAGE_ASSEMBLE"
 	//图片渲染
 	WEBSOCKET_RENDER_IMAGE = "WEBSOCKET_RENDER_IMAGE"
 	//数据格式错误

server/websocket/internal/logic/datatransferlogic.go

@@ -270,9 +270,9 @@ func (w *wsConnectItem) dealwithReciveData(data []byte) {
 	d, _ := json.Marshal(parseInfo.D)
 	//分消息类型给到不同逻辑处理，可扩展
 	switch parseInfo.T {
-	//图片渲染
+	//图片渲染数据组装
-	case constants.WEBSOCKET_RENDER_IMAGE:
+	case constants.WEBSOCKET_RENDER_IMAGE_ASSEMBLE:
-		w.SendToCloudRender(d)
+		w.assembleRenderData(d)
 	default:
 
 	}

server/websocket/internal/logic/ws_render_image_logic.go

@@ -21,8 +21,13 @@ type renderImageControlChanItem struct {
 	RenderId string // map的val
 }
 
-// 渲染请求数据处理发送云渲染服务处理
+// 渲染发送到组装数据组装数据
-func (w *wsConnectItem) SendToCloudRender(data []byte) {
+type assembleRenderData struct {
+	TaskId     string      `json:"task_id"`
+	RenderData interface{} `json:"render_data"`
+}
+
+func (w *wsConnectItem) assembleRenderData(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))
@@ -31,24 +36,23 @@ func (w *wsConnectItem) SendToCloudRender(data []byte) {
 	}
 	logx.Info("收到请求云渲染图片数据：", renderImageData)
 	//把需要渲染的图片任务加进去
-	select {
+	taskId := hash.JsonHashKey(renderImageData.RenderData)
-	case <-w.closeChan: //连接关闭了
+	w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
-		return
+		Option:   1, //0删除 1添加
-	default:
+		TaskId:   taskId,
-		//加入渲染任务
+		RenderId: renderImageData.RenderId,
-		taskId := hash.JsonHashKey(renderImageData.RenderData)
-		w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
-			Option:   1, //0删除 1添加
-			TaskId:   taskId,
-			RenderId: renderImageData.RenderId,
-		}
-		//发送给对应的流水线组装数据
-		if err := w.rabbitMq.SendMsg(constants.RABBIT_MQ_ASSEMBLE_RENDER_DATA, data); err != nil {
-			logx.Error("发送渲染任务数据到MQ失败:", string(data), "err:", err)
-			return
-		}
-		logx.Info("发送渲染数据到rabbitmq成功:", string(data))
 	}
+	tmpData := assembleRenderData{
+		TaskId:     taskId,
+		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))
 }
 
 // 操作连接中渲染任务的增加/删除

server/websocket/websocket.go

@@ -30,11 +30,11 @@ func main() {
 	defer server.Stop()
 
 	ctx := svc.NewServiceContext(c)
-	//消费组装队列
+	//消费渲染结果队列
 	ctx1 := context.Background()
 	ctx2, cancel := context.WithCancel(ctx1)
 	defer cancel()
-	go ctx.RabbitMq.Consume(ctx2, constants.RABBIT_MQ_ASSEMBLE_RENDER_DATA, &consumer.MqConsumerRenderResult{})
+	go ctx.RabbitMq.Consume(ctx2, constants.RABBIT_MQ_RENDER_RESULT_DATA, &consumer.MqConsumerRenderResult{})
 	handler.RegisterHandlers(server, ctx)
 	fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
 	server.Start()