From 9d77536708f86fb5087c0c6149145066ad98153d Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Wed, 15 Oct 2025 14:24:10 +0800
Subject: [PATCH] 增加websocket订阅无人机视频
---
ard-system/src/main/resources/mapper/data/ArdNoFlyZoneMapper.xml | 0
ard-work/src/main/java/com/ard/work/component/websocket/handler/UavStreamManager.java | 131 +++++++++++++++++++++++++++++++++++++++++++
ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java | 42 +++++++++++--
3 files changed, 166 insertions(+), 7 deletions(-)
diff --git a/ard-system/src/main/resources/mapper/noflyzone/ArdNoFlyZoneMapper.xml b/ard-system/src/main/resources/mapper/data/ArdNoFlyZoneMapper.xml
similarity index 100%
rename from ard-system/src/main/resources/mapper/noflyzone/ArdNoFlyZoneMapper.xml
rename to ard-system/src/main/resources/mapper/data/ArdNoFlyZoneMapper.xml
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..f6ce332 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);
@@ -65,6 +67,34 @@
}
} catch (Exception e) {
log.error("Error handling message: {}", message, e);
+ }
+ }
+
+ /**
+ * 处理流消息
+ * <p>
+ * 该方法解析给定的负载数据，以获取无人机流消息，并根据消息类型决定是否开始或停止流
+ *
+ * @param payload 消息的负载数据，通常为JSON字符串
+ * @param userId 接收消息的用户ID
+ */
+ private void handleStreamMessage(String payload, Long userId) {
+ // 解析JSON负载数据为BaseMessage对象，包含UavStreamMessage类型的有效载荷
+ BaseMessage<UavStreamMessage> streamMessage = JSON.parseObject(payload,
+ new TypeReference<BaseMessage<UavStreamMessage>>() {
+ });
+ UavStreamMessage uavStreamMessage = streamMessage.getPayload();
+
+ // 检查解析后的UavStreamMessage对象是否非空
+ if (uavStreamMessage != null) {
+ String boxSn = uavStreamMessage.getBoxSn();
+
+ // 根据Subscribe决定操作类型：true表示开始拉流，false表示停止拉流
+ if (uavStreamMessage.getSubscribe()) {
+ uavStreamManager.subscribe(boxSn, userId);
+ } else {
+ uavStreamManager.unsubscribe(boxSn, userId);
+ }
}
}
@@ -233,11 +263,9 @@
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);
diff --git a/ard-work/src/main/java/com/ard/work/component/websocket/handler/UavStreamManager.java b/ard-work/src/main/java/com/ard/work/component/websocket/handler/UavStreamManager.java
new file mode 100644
index 0000000..e24af6f
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/work/component/websocket/handler/UavStreamManager.java
@@ -0,0 +1,131 @@
+package com.ard.work.component.websocket.handler;
+
+import com.alibaba.fastjson2.JSON;
+
+import com.ard.vtdu.domain.ArdVtdu;
+import com.ard.work.component.websocket.utils.WebSocketUtil;
+import com.ard.work.domain.message.BaseMessage;
+import com.ard.work.domain.receiver.BoxStreamUrlReceiver;
+import com.ard.work.service.ITyService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+@Service
+@Slf4j
+public class UavStreamManager {
+
+ private static final String STREAM_LOG_PREFIX = "【stream】";
+ private static final String STREAM_NOT_FOUND_MSG = "Box stream not found: ";
+ private static final String STREAM_CLOSED_MSG = "Stream closed due to no active users: ";
+ private static final String STREAM_IN_USE_MSG = "Stream is still in use, remaining users: ";
+
+ @Resource
+ private ITyService tyService;
+
+ // 盒子拉流地址缓存 key:boxSn value: webrtcUrl
+ public static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>();
+ // 盒子拉流用户集合缓存 key:boxSn value:Set<UserId>
+ public static final ConcurrentMap<String, Set<Long>> STREAM_USER_MAP = new ConcurrentHashMap<>();
+
+ /**
+ * 开始拉流
+ *
+ * @param boxSn 盒子序列号
+ * @param userId 用户ID
+ */
+ public void subscribe(String boxSn, Long userId) {
+
+ STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
+ String webrtcUrl = getWebrtcUrl(boxSn);
+ if (Objects.equals(webrtcUrl, "")) {
+ log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn);
+ BaseMessage<String> baseMessage = new BaseMessage<>("stream", STREAM_NOT_FOUND_MSG + boxSn);
+ sendMessageToUser(userId, JSON.toJSONString(baseMessage));
+ }
+ BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl);
+ BaseMessage<String> baseMessage = new BaseMessage<>("stream", webrtcUrl);
+ sendMessageToUser(userId, JSON.toJSONString(baseMessage));
+ }
+
+ /**
+ * 停止拉流
+ *
+ * @param boxSn 盒子序列号
+ * @param userId 用户ID
+ */
+ public void unsubscribe(String boxSn, Long userId) {
+ synchronized (this) {
+ Set<Long> users = STREAM_USER_MAP.get(boxSn);
+ if (users == null) {
+ log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, boxSn);
+ BaseMessage<String> baseMessage = new BaseMessage<>("stream", "No active users for box: " + boxSn);
+ sendMessageToUser(userId, JSON.toJSONString(baseMessage));
+ return;
+ }
+
+ users.remove(userId);
+ String message = "";
+ if (users.isEmpty()) {
+ BOX_STREAM_PULL_MAP.remove(boxSn);
+ STREAM_USER_MAP.remove(boxSn);
+ log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
+ message = STREAM_CLOSED_MSG + boxSn;
+ } else {
+ log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
+ message = STREAM_IN_USE_MSG + boxSn;
+ }
+ BaseMessage<String> baseMessage = new BaseMessage<>("stream", message);
+ sendMessageToUser(userId, JSON.toJSONString(baseMessage));
+ }
+ }
+
+ /**
+ * 停止拉流
+ *
+ * @param userId 用户ID
+ */
+ public void stopAllPullStreamByUser(Long userId) {
+ synchronized (this) {
+ STREAM_USER_MAP.forEach((boxSn, users) -> unsubscribe(boxSn, userId));
+ }
+ }
+
+ /**
+ * 发送消息给指定用户
+ *
+ * @param userId 用户ID
+ * @param message 消息内容
+ */
+ private void sendMessageToUser(Long userId, String message) {
+ try {
+ WebSocketUtil.sendMessage(userId, message);
+ } catch (Exception e) {
+ log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e);
+ }
+ }
+
+
+ /**
+ * 通过盒子编号获取webrtc地址
+ *
+ * @param boxSn 盒子编号
+ * @return String webrtcUrl
+ */
+ private String getWebrtcUrl(String boxSn) {
+ try {
+ BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
+ if (boxLiveUrl != null) {
+ return boxLiveUrl.getRtc();
+ }
+ } catch (Exception e) {
+ log.error("获取媒体地址异常：{}", e.getMessage());
+ }
+ return "";
+ }
+}
\ No newline at end of file
--
Gitblit v1.9.3