package com.ard.tianyu.websocket.handler; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.TypeReference; import com.ard.tianyu.service.control.UavControlService; import com.ard.tianyu.service.stream.UavStreamManager; import com.ard.tianyu.websocket.domain.BaseMessage; import com.ard.tianyu.websocket.domain.UavStreamMessage; import com.ard.tianyu.websocket.utils.WebSocketUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import javax.annotation.Resource; @Component @Slf4j public class TyMessageHandler extends TextWebSocketHandler { @Resource private UavStreamManager uavStreamManager; @Resource private UavControlService uavControlService; @Override public void handleTextMessage(WebSocketSession session, TextMessage message) { // 处理接收到的消息 String userIdStr = (String) session.getAttributes().get("userId"); Long userId = Long.parseLong(userIdStr); String payload = message.getPayload(); log.info("【cmd】Received message: {}", payload); log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size()); BaseMessage baseMessage = JSON.parseObject(payload, BaseMessage.class); switch (baseMessage.getType()) { case "stream": BaseMessage streamMessage = JSON.parseObject(payload, new TypeReference>() { }); UavStreamMessage uavStreamMessage = streamMessage.getPayload(); if (uavStreamMessage != null) { String boxSn = uavStreamMessage.getBoxSn(); if (uavStreamMessage.getPullType().equals(1)) { uavStreamManager.startPullStream(boxSn, userId); } else { uavStreamManager.stopPullStream(boxSn, userId); } } break; } } @Override public void afterConnectionEstablished(WebSocketSession session) { log.info("【cmd】Connected: {}", session.getId()); String userIdStr = (String) session.getAttributes().get("userId"); Long userId = Long.parseLong(userIdStr); WebSocketUtil.ONLINE_USER_SESSIONS.put(userId, session); log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size()); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { log.info("【cmd】Disconnected: {}", session.getId()); String userIdStr = (String) session.getAttributes().get("userId"); Long userId = Long.parseLong(userIdStr); WebSocketUtil.ONLINE_USER_SESSIONS.remove(userId); log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size()); //停止用户的所有流 uavStreamManager.stopAllPullStreamByUser(userId); if (userId != null) { // 释放该用户所有 UAV 控制权 uavControlService.releaseAllControls(userId); } } }