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

package logic

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

// 消费渲染结果数据
type MqConsumerRenderResult struct {
}

func (m *MqConsumerRenderResult) Run(ctx context.Context, data []byte) error {
	defer func() {
		if err := recover(); err != nil {
			logx.Error("MqConsumerRenderResult panic:", err)
		}
	}()
	logx.Info("接收到MqConsumerRenderResult数据:", string(data))
	var parseInfo websocket_data.RenderImageNotify
	if err := json.Unmarshal(data, &parseInfo); err != nil {
		logx.Error("MqConsumerRenderResult data format err:", err)
		return nil //不返回错误则就删掉该消息
	}
	//遍历websocket链接把数据传进去
	mapConnPool.Range(func(key, value any) bool {
		//断言连接
		ws, ok := value.(wsConnectItem)
		if !ok {
			return true
		}
		//关闭标识
		if ws.isClose {
			return true
		}
		//查询有无该渲染任务
		renderId, ok := ws.renderProperty.renderImageTask[parseInfo.TaskId]
		if !ok {
			return true
		}
		b := ws.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
			RenderId: renderId,
			Image:    parseInfo.Image,
		})
		//删除对应的需要渲染的图片map
		ws.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
			Option:   0, //0删除 1添加
			TaskId:   parseInfo.TaskId,
			RenderId: renderId,
		}
		//发送数据到out chan
		ws.sendToOutChan(b)
		return true
	})
	return nil
}