| | |
| | | import org.springframework.web.socket.handler.TextWebSocketHandler; |
| | | |
| | | import javax.annotation.Resource; |
| | | import java.util.Map; |
| | | import java.util.Objects; |
| | | import java.util.concurrent.CompletableFuture; |
| | | import java.util.concurrent.ConcurrentHashMap; |
| | | |
| | | |
| | | @Component |
| | |
| | | @Resource |
| | | private UavAllManager uavAllManager; |
| | | @Resource |
| | | private NoFlyDistanceManager noFlyDistanceManager; |
| | | @Resource |
| | | private TokenService tokenService; |
| | | private final Map<Long, Long> lastDisconnectTime = new ConcurrentHashMap<>(); |
| | | |
| | | @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: {}", WebSocketSessionManager.getOnlineUserCount()); |
| | | log.info("【ws】【{}】Received message: {}, size: {}", userId, payload, |
| | | WebSocketSessionManager.getOnlineUserCount()); |
| | | BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class); |
| | | MessageType messageType = MessageType.fromString(baseMessage.getType()); |
| | | switch (messageType) { |
| | |
| | | break; |
| | | case CONTROLLER: |
| | | handleControlMessage(payload, userId); |
| | | break; |
| | | case NO_FLY_DISTANCE: |
| | | handleNoFlyDistanceMessage(payload, userId); |
| | | break; |
| | | case ALL: |
| | | handleAllMessage(payload, userId); |
| | |
| | | } |
| | | } |
| | | |
| | | /** |
| | | * 处理禁飞区距离 |
| | | * |
| | | * @param payload 消息负载,包含订阅信息 |
| | | * @param userId 用户ID,用于标识用户 |
| | | */ |
| | | private void handleNoFlyDistanceMessage(String payload, Long userId) { |
| | | // 解析接收到的JSON消息,将其转换为BaseMessage对象 |
| | | BaseMessage<NoFlyDistanceMessage> baseMessage = JSON.parseObject(payload, |
| | | new TypeReference<BaseMessage<NoFlyDistanceMessage>>() { |
| | | }); |
| | | // 获取消息的有效载荷部分 |
| | | NoFlyDistanceMessage noFlyDistanceMessage = baseMessage.getPayload(); |
| | | |
| | | // 检查有效载荷是否存在 |
| | | if (noFlyDistanceMessage != null) { |
| | | // 获取无人机盒子的序列号 |
| | | String boxSn = noFlyDistanceMessage.getBoxSn(); |
| | | // 根据消息中的订阅标志,决定是订阅还是退订无人机状态 |
| | | Boolean subscribe = noFlyDistanceMessage.getSubscribe(); |
| | | if (subscribe) { |
| | | // 执行订阅操作 |
| | | noFlyDistanceManager.subscribe(boxSn, userId, 1); |
| | | } else { |
| | | // 执行退订操作 |
| | | noFlyDistanceManager.unsubscribe(boxSn, userId); |
| | | } |
| | | } |
| | | } |
| | | |
| | | |
| | | @Override |
| | | public void afterConnectionEstablished(WebSocketSession session) { |
| | | log.info("【cmd】Connected: {}", session.getId()); |
| | | String userIdStr = (String) session.getAttributes().get("userId"); |
| | | if (userIdStr == null || userIdStr.isBlank()) { |
| | | return; |
| | | } |
| | | Long userId = Long.parseLong(userIdStr); |
| | | lastDisconnectTime.remove(userId); // 清除断开标记 |
| | | WebSocketSessionManager.addSession(userId, session); |
| | | log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount()); |
| | | log.info("【ws】【{}】Connected,size: {}",userId, WebSocketSessionManager.getOnlineUserCount()); |
| | | } |
| | | |
| | | @Override |
| | | public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { |
| | | log.info("【cmd】Disconnected: {}", session.getId()); |
| | | |
| | | String userIdStr = (String) session.getAttributes().get("userId"); |
| | | if (userIdStr == null || userIdStr.isBlank()) { |
| | | return; |
| | | } |
| | | Long userId = Long.parseLong(userIdStr); |
| | | WebSocketSessionManager.removeSession(userId); |
| | | log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount()); |
| | | log.info("【ws】【{}】Disconnected,size: {}",userId, WebSocketSessionManager.getOnlineUserCount()); |
| | | long version = System.currentTimeMillis(); |
| | | lastDisconnectTime.put(userId, version); |
| | | // 释放该用户所有 UAV 控制权 |
| | | |
| | | // 延迟10秒删除token(若期间没重连) |
| | | CompletableFuture.runAsync(() -> { |
| | | try { |
| | | Thread.sleep(10_000); // 延迟10秒 |
| | | Thread.sleep(5_000); // 延迟5秒 |
| | | } catch (InterruptedException ignored) { |
| | | } |
| | | |
| | | // 如果版本号不一致,说明用户曾重连 → 不执行删除 |
| | | if (!Objects.equals(lastDisconnectTime.get(userId), version)) { |
| | | return; |
| | | } |
| | | if (!WebSocketSessionManager.containsUser(userId)) { |
| | | uavControlService.releaseAllControls(userId); |
| | | uavStreamManager.stopAllPullStreamByUser(userId); |
| | | uavStateManager.removeSubscribe(userId); |
| | | uavTelemetryManager.removeSubscribe(userId); |
| | | uavSignalManager.removeSubscribe(userId); |
| | | uavControlManager.removeSubscribe(userId); |
| | | LoginUser loginUser = tokenService.getLoginUserByUserId(userId); |
| | | if (loginUser != null) { |
| | | uavControlService.releaseAllControls(userId); |
| | | uavStreamManager.stopAllPullStreamByUser(userId); |
| | | uavStateManager.removeSubscribe(userId); |
| | | uavTelemetryManager.removeSubscribe(userId); |
| | | uavSignalManager.removeSubscribe(userId); |
| | | uavControlManager.removeSubscribe(userId); |
| | | tokenService.delLoginUser(loginUser.getToken(), userId); |
| | | log.info("【cmd】用户 [{}] 超时未重连,已删除 token", userId); |
| | | } |