From 8e4e1c1cfd18ff0b89c2e8c7005f7e55a5aed591 Mon Sep 17 00:00:00 2001
From: laodaming <11058467+laudamine@user.noreply.gitee.com>
Date: Mon, 7 Aug 2023 11:15:51 +0800
Subject: [PATCH] fix

---
 constants/websocket.go                        |  2 +
 .../internal/logic/datatransferlogic.go       |  6 +--
 .../internal/logic/ws_render_image_logic.go   | 42 ++++++++++---------
 server/websocket/websocket.go                 |  4 +-
 4 files changed, 30 insertions(+), 24 deletions(-)

diff --git a/constants/websocket.go b/constants/websocket.go
index 34e28eb7..be28eb32 100644
--- a/constants/websocket.go
+++ b/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"
 	//数据格式错误
diff --git a/server/websocket/internal/logic/datatransferlogic.go b/server/websocket/internal/logic/datatransferlogic.go
index 317a93ad..1865300d 100644
--- a/server/websocket/internal/logic/datatransferlogic.go
+++ b/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:
-		w.SendToCloudRender(d)
+	//图片渲染数据组装
+	case constants.WEBSOCKET_RENDER_IMAGE_ASSEMBLE:
+		w.assembleRenderData(d)
 	default:
 
 	}
diff --git a/server/websocket/internal/logic/ws_render_image_logic.go b/server/websocket/internal/logic/ws_render_image_logic.go
index 3ba3fdd7..ca93a6c7 100644
--- a/server/websocket/internal/logic/ws_render_image_logic.go
+++ b/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 {
-	case <-w.closeChan: //连接关闭了
-		return
-	default:
-		//加入渲染任务
-		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))
+	taskId := hash.JsonHashKey(renderImageData.RenderData)
+	w.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
+		Option:   1, //0删除 1添加
+		TaskId:   taskId,
+		RenderId: renderImageData.RenderId,
 	}
+	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))
 }
 
 // 操作连接中渲染任务的增加/删除
diff --git a/server/websocket/websocket.go b/server/websocket/websocket.go
index c62fc5ba..01a9efb8 100644
--- a/server/websocket/websocket.go
+++ b/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()