From f0ed143646e3c280416ca05f64c1b59604289928 Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Mon, 20 Oct 2025 10:52:48 +0800
Subject: [PATCH] websocket无人机控制者和视频流数据增加boxSn

---
 ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java |   82 +++++++++++++++++++++++++++++-----------
 1 files changed, 59 insertions(+), 23 deletions(-)

diff --git a/ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java b/ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java
index 58553f5..b7e380a 100644
--- a/ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java
+++ b/ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java
@@ -23,6 +23,8 @@
     @Resource
     private UavControlService uavControlService;
     @Resource
+    private UavStreamManager uavStreamManager;
+    @Resource
     private UavStateManager uavStateManager;
     @Resource
     private UavTelemetryManager uavTelemetryManager;
@@ -44,8 +46,8 @@
             BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class);
             MessageType messageType = MessageType.fromString(baseMessage.getType());
             switch (messageType) {
-                case STREAM: // 直播地址
-                    
+                case STREAM:
+                    handleStreamMessage(payload, userId);
                     break;
                 case STATE:
                     handleStateMessage(payload, userId);
@@ -69,6 +71,34 @@
     }
 
     /**
+     * 处理流消息
+     * <p>
+     * 该方法解析给定的负载数据,以获取无人机流消息,并根据消息类型决定是否开始或停止流
+     *
+     * @param payload 消息的负载数据,通常为JSON字符串
+     * @param userId  接收消息的用户ID
+     */
+    private void handleStreamMessage(String payload, Long userId) {
+        // 解析JSON负载数据为BaseMessage对象,包含UavStreamMessage类型的有效载荷
+        BaseMessage<UavStreamMessage> baseMessage = JSON.parseObject(payload,
+                new TypeReference<BaseMessage<UavStreamMessage>>() {
+                });
+        UavStreamMessage uavStreamMessage = baseMessage.getPayload();
+
+        // 检查解析后的UavStreamMessage对象是否非空
+        if (uavStreamMessage != null) {
+            String boxSn = uavStreamMessage.getBoxSn();
+
+            // 根据Subscribe决定操作类型:true表示开始拉流,false表示停止拉流
+            if (uavStreamMessage.getSubscribe()) {
+                uavStreamManager.subscribe(boxSn, userId);
+            } else {
+                uavStreamManager.unsubscribe(boxSn, userId);
+            }
+        }
+    }
+
+    /**
      * 处理状态消息
      * 根据接收到的消息内容,决定无人机状态信息的订阅或退订
      *
@@ -77,11 +107,11 @@
      */
     private void handleStateMessage(String payload, Long userId) {
         // 解析接收到的JSON消息,将其转换为BaseMessage对象
-        BaseMessage<UavStateMessage> stateMessage = JSON.parseObject(payload,
+        BaseMessage<UavStateMessage> baseMessage = JSON.parseObject(payload,
                 new TypeReference<BaseMessage<UavStateMessage>>() {
                 });
         // 获取消息的有效载荷部分
-        UavStateMessage uavStateMessage = stateMessage.getPayload();
+        UavStateMessage uavStateMessage = baseMessage.getPayload();
 
         // 检查有效载荷是否存在
         if (uavStateMessage != null) {
@@ -108,11 +138,11 @@
      */
     private void handleTelemetryMessage(String payload, Long userId) {
         // 解析接收到的JSON消息,将其转换为BaseMessage对象
-        BaseMessage<UavTelemetryMessage> telemetryMessage = JSON.parseObject(payload,
+        BaseMessage<UavTelemetryMessage> baseMessage = JSON.parseObject(payload,
                 new TypeReference<BaseMessage<UavTelemetryMessage>>() {
                 });
         // 获取消息的有效载荷部分
-        UavTelemetryMessage uavTelemetryMessage = telemetryMessage.getPayload();
+        UavTelemetryMessage uavTelemetryMessage = baseMessage.getPayload();
 
         // 检查有效载荷是否存在
         if (uavTelemetryMessage != null) {
@@ -139,11 +169,11 @@
      */
     private void handleSignalMessage(String payload, Long userId) {
         // 解析接收到的JSON消息,将其转换为BaseMessage对象
-        BaseMessage<UavSignalMessage> signalMessage = JSON.parseObject(payload,
+        BaseMessage<UavSignalMessage> baseMessage = JSON.parseObject(payload,
                 new TypeReference<BaseMessage<UavSignalMessage>>() {
                 });
         // 获取消息的有效载荷部分
-        UavSignalMessage uavSignalMessage = signalMessage.getPayload();
+        UavSignalMessage uavSignalMessage = baseMessage.getPayload();
 
         // 检查有效载荷是否存在
         if (uavSignalMessage != null) {
@@ -162,19 +192,19 @@
     }
 
     /**
-     * 处理无人机控制
-     * 根据接收到的消息内容,决定无人机控制信息的订阅或退订
+     * 处理无人机控制者
+     * 根据接收到的消息内容,决定无人机控制者信息的订阅或退订
      *
      * @param payload 消息负载,包含控制信息
      * @param userId  用户ID,用于标识用户
      */
     private void handleControlMessage(String payload, Long userId) {
         // 解析接收到的JSON消息,将其转换为BaseMessage对象
-        BaseMessage<UavControlMessage> controlMessage = JSON.parseObject(payload,
+        BaseMessage<UavControlMessage> baseMessage = JSON.parseObject(payload,
                 new TypeReference<BaseMessage<UavControlMessage>>() {
                 });
         // 获取消息的有效载荷部分
-        UavControlMessage uavControlMessage = controlMessage.getPayload();
+        UavControlMessage uavControlMessage = baseMessage.getPayload();
 
         // 检查有效载荷是否存在
         if (uavControlMessage != null) {
@@ -194,18 +224,18 @@
 
     private void handleAllMessage(String payload, Long userId) {
         // 解析接收到的JSON消息,将其转换为BaseMessage对象
-        BaseMessage<UavControlMessage> controlMessage = JSON.parseObject(payload,
-                new TypeReference<BaseMessage<UavControlMessage>>() {
+        BaseMessage<UavAllMessage> baseMessage = JSON.parseObject(payload,
+                new TypeReference<BaseMessage<UavAllMessage>>() {
                 });
         // 获取消息的有效载荷部分
-        UavControlMessage uavControlMessage = controlMessage.getPayload();
+        UavAllMessage uavAllMessage = baseMessage.getPayload();
 
         // 检查有效载荷是否存在
-        if (uavControlMessage != null) {
+        if (uavAllMessage != null) {
             // 获取无人机盒子的序列号
-            String boxSn = uavControlMessage.getBoxSn();
+            String boxSn = uavAllMessage.getBoxSn();
             // 根据消息中的订阅标志,决定是订阅还是退订无人机状态
-            Boolean subscribe = uavControlMessage.getSubscribe();
+            Boolean subscribe = uavAllMessage.getSubscribe();
             if (subscribe) {
                 // 执行订阅操作
                 uavAllManager.subscribe(boxSn, userId, 1);
@@ -220,6 +250,9 @@
     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);
         WebSocketSessionManager.addSession(userId, session);
         log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount());
@@ -229,18 +262,21 @@
     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());
 
-        if (userId != null) {
-            // 释放该用户所有 UAV 控制权
-            uavControlService.releaseAllControls(userId);
-        }
-
+        // 释放该用户所有 UAV 控制权
+        uavControlService.releaseAllControls(userId);
+        uavStreamManager.stopAllPullStreamByUser(userId);
         uavStateManager.removeSubscribe(userId);
         uavTelemetryManager.removeSubscribe(userId);
         uavSignalManager.removeSubscribe(userId);
         uavControlManager.removeSubscribe(userId);
     }
+
+
 }

--
Gitblit v1.9.3