From de7cddcd9a5597ac281fbd619f4f0bba09772386 Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Tue, 11 Mar 2025 16:19:35 +0800
Subject: [PATCH] 优化遥测、信道治理、状态 websocket
---
ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStreamManager.java | 31 ++++++++++++++++++++-----------
1 files changed, 20 insertions(+), 11 deletions(-)
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStreamManager.java b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStreamManager.java
index 308b47a..7e21147 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStreamManager.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStreamManager.java
@@ -1,7 +1,10 @@
package com.ard.tianyu.websocket.handler;
+import com.alibaba.fastjson2.JSON;
+import com.ard.tianyu.domain.dto.state.UavState;
import com.ard.tianyu.domain.receiver.BoxStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
+import com.ard.tianyu.websocket.domain.BaseMessage;
import com.ard.tianyu.websocket.utils.WebSocketUtil;
import com.ard.vtdu.domain.ArdVtdu;
import com.ard.vtdu.service.IMediaService;
@@ -19,9 +22,9 @@
public class UavStreamManager {
private static final String STREAM_LOG_PREFIX = "【stream】";
- private static final String STREAM_NOT_FOUND_MSG = STREAM_LOG_PREFIX + "Box stream not found: ";
- private static final String STREAM_CLOSED_MSG = STREAM_LOG_PREFIX + "Stream closed due to no active users: ";
- private static final String STREAM_IN_USE_MSG = STREAM_LOG_PREFIX + "Stream is still in use, remaining users: ";
+ 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;
@@ -39,7 +42,7 @@
* @param boxSn 盒子序列号
* @param userId 用户ID
*/
- public void startPullStream(String boxSn, Long userId) {
+ public void subscribe(String boxSn, Long userId) {
String webrtcUrl = "";
STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
//不管谁拉流都判断流媒体中是否存在
@@ -48,13 +51,15 @@
webrtcUrl = addVtdu(boxSn);
if (Objects.equals(webrtcUrl, "")) {
log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn);
- sendMessageToUser(userId, 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);
} else {
webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn);
}
- sendMessageToUser(userId, webrtcUrl);
+ BaseMessage<String> baseMessage = new BaseMessage<>("stream", webrtcUrl);
+ sendMessageToUser(userId, JSON.toJSONString(baseMessage));
}
/**
@@ -63,26 +68,30 @@
* @param boxSn 盒子序列号
* @param userId 用户ID
*/
- public void stopPullStream(String boxSn, Long userId) {
+ 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);
- sendMessageToUser(userId, "No active users for box: " + 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()) {
mediaService.removePath(boxSn);
BOX_STREAM_PULL_MAP.remove(boxSn);
STREAM_USER_MAP.remove(boxSn);
log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
- sendMessageToUser(userId, STREAM_CLOSED_MSG + boxSn);
+ message = STREAM_CLOSED_MSG + boxSn;
} else {
log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
- sendMessageToUser(userId, STREAM_IN_USE_MSG + users);
+ message = STREAM_IN_USE_MSG + boxSn;
}
+ BaseMessage<String> baseMessage = new BaseMessage<>("stream", message);
+ sendMessageToUser(userId, JSON.toJSONString(baseMessage));
}
}
@@ -93,7 +102,7 @@
*/
public void stopAllPullStreamByUser(Long userId) {
synchronized (this) {
- STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
+ STREAM_USER_MAP.forEach((boxSn, users) -> unsubscribe(boxSn, userId));
}
}
--
Gitblit v1.9.3