package logic import ( "bytes" "encoding/json" "fmt" "fusenapi/constants" "fusenapi/server/websocket/internal/types" "github.com/google/uuid" "github.com/gorilla/websocket" "github.com/zeromicro/go-zero/rest/httpx" "net/http" "sync" "time" "context" "fusenapi/server/websocket/internal/svc" "github.com/zeromicro/go-zero/core/logx" ) type DataTransferLogic struct { logx.Logger ctx context.Context svcCtx *svc.ServiceContext } func NewDataTransferLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DataTransferLogic { return &DataTransferLogic{ Logger: logx.WithContext(ctx), ctx: ctx, svcCtx: svcCtx, } } var ( //临时缓存对象池 buffPool = sync.Pool{ New: func() interface{} { return bytes.Buffer{} }, } //升级websocket upgrade = websocket.Upgrader{ //握手超时时间15s HandshakeTimeout: time.Second * 15, //允许跨域 CheckOrigin: func(r *http.Request) bool { return true }, Error: func(w http.ResponseWriter, r *http.Request, status int, reason error) { httpx.ErrorCtx(r.Context(), w, reason) return }, WriteBufferPool: &buffPool, } //连接map池 mapConnPool = sync.Map{} ) // 每个连接的连接基本属性 type wsConnectItem struct { conn *websocket.Conn //websocket的连接 closeChan chan struct{} //ws连接关闭chan isClose bool //是否已经关闭 flag string //ws连接唯一标识 inChan chan []byte //接受消息缓冲通道 outChan chan []byte //发送回客户端的消息 mutex sync.Mutex //互斥锁 renderProperty renderProperty //扩展云渲染属性 } func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.ResponseWriter, r *http.Request) { //升级websocket conn, err := upgrade.Upgrade(w, r, nil) if err != nil { logx.Error("http upgrade websocket err:", err) return } defer conn.Close() rsp := types.DataTransferData{} /*isAuth := true // 解析JWT token,并对空用户进行判断 claims, err := svcCtx.ParseJwtToken(r) // 如果解析JWT token出错,则返回未授权的JSON响应并记录错误消息 if err != nil { rsp.T = constants.WEBSOCKET_UNAUTH rsp.D = "unAuth" b, _ := json.Marshal(rsp) _ = conn.WriteMessage(websocket.TextMessage, b) isAuth = false } if claims != nil { // 从token中获取对应的用户信息 _, err = auth.GetUserInfoFormMapClaims(claims) // 如果获取用户信息出错,则返回未授权的JSON响应并记录错误消息 if err != nil { rsp.T = constants.WEBSOCKET_UNAUTH rsp.D = "unAuth!!" b, _ := json.Marshal(rsp) _ = conn.WriteMessage(websocket.TextMessage, b) isAuth = false } } else { // 如果claims为nil,则认为用户身份为白板用户 rsp.T = constants.WEBSOCKET_UNAUTH rsp.D = "unAuth!!!" b, _ := json.Marshal(rsp) _ = conn.WriteMessage(websocket.TextMessage, b) isAuth = false } //不是授权的连接(10秒后关闭) if !isAuth { select { case <-time.After(time.Second * 10): conn.Close() return } }*/ //生成连接唯一标识 flag := uuid.New().String() + time.Now().Format("20060102150405") ws := wsConnectItem{ conn: conn, flag: flag, closeChan: make(chan struct{}, 1), inChan: make(chan []byte, 100), outChan: make(chan []byte, 100), renderProperty: renderProperty{ renderImageTask: make(map[string]struct{}), renderImageTaskCtlChan: make(chan renderImageControlChanItem, 100), }, } //保存连接 mapConnPool.Store(flag, ws) defer ws.close() //把连接成功消息发回去 rsp.T = constants.WEBSOCKET_CONNECT_SUCCESS rsp.D = flag b, _ := json.Marshal(rsp) conn.WriteMessage(websocket.TextMessage, b) //循环读客户端信息 go ws.readLoop() //循环把数据发送给客户端 go ws.writeLoop() //推消息到云渲染 go ws.sendLoop() //操作连接中渲染任务的增加/删除 go ws.operationRenderTask() //心跳 ws.heartbeat() } // 心跳 func (w *wsConnectItem) heartbeat() { rsp := types.DataTransferData{ T: constants.WEBSOCKET_HEARTBEAT, D: "", } b, _ := json.Marshal(rsp) for { time.Sleep(time.Second * 10) select { case <-w.closeChan: return default: //发送心跳信息 if err := w.conn.WriteMessage(websocket.TextMessage, b); err != nil { logx.Error("发送心跳信息异常,关闭连接:", w.flag, err) w.close() return } } } } // 关闭连接 func (w *wsConnectItem) close() { w.mutex.Lock() defer w.mutex.Unlock() logx.Info("websocket:", w.flag, " is closing...") w.conn.Close() mapConnPool.Delete(w.flag) if !w.isClose { w.isClose = true close(w.closeChan) } logx.Info("websocket:", w.flag, " is closed") } // 读取输出返回给客户端 func (w *wsConnectItem) writeLoop() { for { select { case <-w.closeChan: //如果关闭了 return case data := <-w.outChan: w.conn.WriteMessage(websocket.TextMessage, data) } } } // 接受客户端发来的消息 func (w *wsConnectItem) readLoop() { for { select { case <-w.closeChan: //如果关闭了 return default: _, data, err := w.conn.ReadMessage() if err != nil { logx.Error("接受信息错误:", err) //关闭连接 w.close() return } //消息传入缓冲通道 w.inChan <- data } } } // 把收到的消息发往不同的地方处理 func (w *wsConnectItem) sendLoop() { for { select { case <-w.closeChan: return case data := <-w.inChan: w.dealwithReciveData(data) } } } // 把要传递给客户端的数据放入outchan func (w *wsConnectItem) sendToOutChan(data []byte) { select { case <-w.closeChan: return case w.outChan <- data: logx.Info("notify send render result to out chan") } } // 获取需要渲染图片的map key func (w *wsConnectItem) getRenderImageMapKey(productId, sizeId, templateId int64) string { return fmt.Sprintf("%d-%d-%d", productId, sizeId, templateId) } // 处理接受到的数据 func (w *wsConnectItem) dealwithReciveData(data []byte) { var parseInfo types.DataTransferData if err := json.Unmarshal(data, &parseInfo); err != nil { logx.Error("invalid format of websocket message") return } d, _ := json.Marshal(parseInfo.D) //分消息类型给到不同逻辑处理,可扩展 switch parseInfo.T { //图片渲染 case constants.WEBSOCKET_RENDER_IMAGE: go w.SendToCloudRender(d) default: } }