From 8ccdc1da5a12248af84383b92c1bcd1ea921b71f Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Sat, 06 Dec 2025 15:27:33 +0800
Subject: [PATCH] 2025-12-06 zwy 在机场设备详情页中，设置机场摄像头舱内、舱外，飞行器视频广角、变焦、红外时，将状态广播发送所有在线用户更改页面按钮选项；
---
ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveCapacityHandler.java | 3
ard-admin/src/main/java/com/ard/web/controller/sample/DeviceController.java | 1
ard-sample/src/main/java/com/ard/sample/manage/service/IDeviceService.java | 3
ard-admin/src/main/java/com/ard/web/controller/sample/LiveStreamController.java | 1
ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java | 1
ard-common/src/main/java/com/ard/common/core/redis/RedisConst.java | 3
ard-sample/src/main/java/com/ard/sample/component/mqtt/model/StateDataEnum.java | 7
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 425 ++++++++++++++++++++++----------------
ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java | 13 +
ard-sample/src/main/java/com/ard/sample/manage/service/impl/LiveStreamServiceImpl.java | 9
ard-sample/src/main/java/com/ard/sample/manage/model/dto/DeviceDTO.java | 2
ard-sample/src/main/java/com/ard/sample/manage/model/receiver/OsdGatewayReceiver.java | 1
ard-sample/src/main/java/com/ard/sample/manage/model/receiver/LiveStatusTwoReceiver.java | 37 +++
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java | 2
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java | 67 +++++
ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveStatusHandler.java | 65 +++++
ard-sample/src/main/java/com/ard/sample/manage/model/entity/DeviceEntity.java | 3
17 files changed, 452 insertions(+), 191 deletions(-)
diff --git a/ard-admin/src/main/java/com/ard/web/controller/sample/DeviceController.java b/ard-admin/src/main/java/com/ard/web/controller/sample/DeviceController.java
index 291caac..f27c889 100644
--- a/ard-admin/src/main/java/com/ard/web/controller/sample/DeviceController.java
+++ b/ard-admin/src/main/java/com/ard/web/controller/sample/DeviceController.java
@@ -265,6 +265,7 @@
.workType(dto.getWorkType())
.webrtcUrl(dto.getWebrtcUrl())
.cameraWebrtcUrl(dto.getCameraWebrtcUrl())
+ .cameraState(dto.getCameraState())
.compatibleStatus(dto.getFirmwareStatus() == null ? null :
DeviceFirmwareStatusEnum.CONSISTENT_UPGRADE != DeviceFirmwareStatusEnum.find(dto.getFirmwareStatus()))
.build();
diff --git a/ard-admin/src/main/java/com/ard/web/controller/sample/LiveStreamController.java b/ard-admin/src/main/java/com/ard/web/controller/sample/LiveStreamController.java
index b3d901a..ace530f 100644
--- a/ard-admin/src/main/java/com/ard/web/controller/sample/LiveStreamController.java
+++ b/ard-admin/src/main/java/com/ard/web/controller/sample/LiveStreamController.java
@@ -9,6 +9,7 @@
import com.ard.sample.manage.model.dto.LiveTypeDTO;
import com.ard.sample.manage.model.param.RemoteDebugParam;
import com.ard.sample.manage.model.receiver.LiveCapacityReceiver;
+import com.ard.sample.manage.service.IDeviceService;
import com.ard.sample.manage.service.ILiveStreamService;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.swagger.annotations.Api;
diff --git a/ard-common/src/main/java/com/ard/common/core/redis/RedisConst.java b/ard-common/src/main/java/com/ard/common/core/redis/RedisConst.java
index e1101c7..7d44b67 100644
--- a/ard-common/src/main/java/com/ard/common/core/redis/RedisConst.java
+++ b/ard-common/src/main/java/com/ard/common/core/redis/RedisConst.java
@@ -21,6 +21,9 @@
public static final String WEBSOCKET_VIDEO_ADDRESS_PREFIX = "videoaddress" + DELIMITER;
+ // cover_state
+ public static final String COVER_STATE_PREFIX = "coverstate" + DELIMITER;
+
public static final String DEVICE_ONLINE_PREFIX = "online" + DELIMITER;
public static final String WEBSOCKET_PREFIX = "webSocket" + DELIMITER;
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveCapacityHandler.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveCapacityHandler.java
index 916482f..d612bb8 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveCapacityHandler.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveCapacityHandler.java
@@ -20,7 +20,7 @@
@Slf4j
public class StateLiveCapacityHandler extends AbstractStateTopicHandler {
- protected StateLiveCapacityHandler(@Autowired @Qualifier("stateFirmwareVersionHandler") AbstractStateTopicHandler handler) {
+ protected StateLiveCapacityHandler(@Autowired @Qualifier("stateLiveStatusHandler") AbstractStateTopicHandler handler) {
super(handler);
}
@@ -28,6 +28,7 @@
public CommonTopicReceiver handleState(Map<String, Object> dataNode, CommonTopicReceiver stateReceiver, String sn) throws JsonProcessingException {
// Determine if it is live capacity data based on name.
if (dataNode.containsKey(StateDataEnum.LIVE_CAPACITY.getDesc())) {
+ Object data = dataNode.get(StateDataEnum.LIVE_CAPACITY.getDesc());
stateReceiver.setData(mapper.convertValue(
dataNode.get(StateDataEnum.LIVE_CAPACITY.getDesc()),
LiveCapacityReceiver.class));
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveStatusHandler.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveStatusHandler.java
new file mode 100644
index 0000000..2227171
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/StateLiveStatusHandler.java
@@ -0,0 +1,65 @@
+package com.ard.sample.component.mqtt.handler;
+
+import com.ard.sample.component.mqtt.model.CommonTopicReceiver;
+import com.ard.sample.component.mqtt.model.StateDataEnum;
+import com.ard.sample.component.websocket.enums.MessageType;
+import com.ard.sample.component.websocket.handler.DockWebSocketHandler;
+import com.ard.sample.manage.model.dto.DeviceDTO;
+import com.ard.sample.manage.model.param.DeviceQueryParam;
+import com.ard.sample.manage.model.receiver.LiveCapacityReceiver;
+import com.ard.sample.manage.model.receiver.LiveStatusTwoReceiver;
+import com.ard.sample.manage.service.IDeviceService;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.Map;
+
+@Service
+@Slf4j
+public class StateLiveStatusHandler extends AbstractStateTopicHandler{
+
+ @Autowired
+ private DockWebSocketHandler _dockWebSocketHandler;
+ @Autowired
+ private IDeviceService _deviceService;
+
+ protected StateLiveStatusHandler(@Autowired @Qualifier("stateFirmwareVersionHandler")AbstractStateTopicHandler handler) {
+ super(handler);
+ }
+
+ @Override
+ public CommonTopicReceiver handleState(Map<String, Object> dataNode, CommonTopicReceiver stateReceiver, String sn) throws JsonProcessingException {
+ if (dataNode.containsKey(StateDataEnum.LIVESTATUS.getDesc())) {
+ stateReceiver.setType(MessageType.LIVESTATUS.getType());
+ String deviceSn= stateReceiver.getGateway();
+ DeviceQueryParam param = DeviceQueryParam.builder().deviceSn(deviceSn).build();
+ List<DeviceDTO> deviceDTOS = _deviceService.getDevicesByParams(param);
+ // List<DeviceDTO> gatewaysList = _deviceService.getDevicesByParams(DeviceQueryParam.builder().childSn(deviceSn).build());
+ String childSn = deviceDTOS.get(0).getChildDeviceSn();
+ List<LiveStatusTwoReceiver> list = mapper.convertValue(dataNode.get(StateDataEnum.LIVESTATUS.getDesc()), new TypeReference<List<LiveStatusTwoReceiver>>() {});
+ for (LiveStatusTwoReceiver receiver : list) {
+ if(receiver.getVideoId().contains(childSn)){
+ stateReceiver.setData(receiver);
+ continue;
+ }
+ }
+ stateReceiver.setGateway(childSn);
+ String liveStatus = "";
+ try{
+ liveStatus=mapper.writeValueAsString(stateReceiver);
+ }
+ catch (Exception e){
+ log.error("接收state直播设备状态反馈转换异常:{}",e);
+ }
+ _deviceService.setChildRedisCamera(childSn, liveStatus);
+ log.info("接收state直播设备状态反馈:{}",liveStatus);
+ return stateReceiver;
+ }
+ return handler.handleState(dataNode, stateReceiver, sn);
+ }
+}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/StateDataEnum.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/StateDataEnum.java
index 715918c..f3024c3 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/StateDataEnum.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/StateDataEnum.java
@@ -12,7 +12,12 @@
LIVE_CAPACITY("live_capacity"),
- PAYLOADS("payloads");
+ PAYLOADS("payloads"),
+
+ /**
+ * 网关当前整体直播状态推送
+ * */
+ LIVESTATUS("live_status");
private String desc;
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java
index a687ad0..4f08004 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java
@@ -6,6 +6,7 @@
CONTROL("control"),
PROGRESS("progress"),
LIVESTREAM("livestream"),
+ LIVESTATUS("livestatus"),
PING("ping"), // 新增心跳请求
PONG("pong"), // 新增心跳响应
ALL("all");
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
index 36fc65c..51c8a5a 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
@@ -1,7 +1,6 @@
package com.ard.sample.component.websocket.handler;
import com.alibaba.fastjson2.JSON;
-import com.ard.sample.common.model.ResponseResult;
import com.ard.sample.component.websocket.enums.MessageType;
import com.ard.sample.component.websocket.model.Payload;
import com.ard.sample.component.websocket.model.WebSocketMessage;
@@ -14,6 +13,7 @@
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
+import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
@@ -21,6 +21,10 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+/**
+ * WebSocket处理器（设备对接）
+ * 核心能力：心跳检测、订阅/取消订阅、定向消息推送、广播消息
+ */
@Slf4j
@Component
public class DockWebSocketHandler extends TextWebSocketHandler {
@@ -31,108 +35,143 @@
@Autowired
private IControlService _controlService;
- // ====================== 心跳相关常量 ======================
- /** 心跳间隔（秒）：客户端需每隔此时间发一次ping */
+ // ====================== 心跳相关常量（可配置化建议放到配置文件） ======================
+ /** 心跳间隔（秒）：客户端需每隔此时间发送一次PING消息 */
private static final long HEARTBEAT_INTERVAL = 30;
- /** 心跳超时阈值（秒）：超过此时间没收到ping，判定连接失效 */
+ /** 心跳超时阈值（秒）：超过此时间未收到PING则判定连接失效 */
private static final long HEARTBEAT_TIMEOUT = 90;
- /** 心跳检测任务间隔（秒）：定时检查超时连接 */
- private static final long CHECK_INTERVAL = 10;
+ /** 心跳检测任务间隔（秒）：定时扫描超时连接 */
+ private static final long HEARTBEAT_CHECK_INTERVAL = 10;
- // ====================== 核心存储 ======================
- // 存储用户会话：userId -> WebSocketSession
+ // ====================== 核心存储（线程安全） ======================
+ /** 用户会话映射：userId -> WebSocketSession */
private final Map<String, WebSocketSession> userSessions = new ConcurrentHashMap<>();
- // 订阅关系：key=订阅标识(sn+type)，value=订阅用户ID集合
+ /** 订阅关系映射：订阅标识(sn:type) -> 订阅用户ID集合 */
private final Map<String, Set<String>> subscriptions = new ConcurrentHashMap<>();
- // 关键：为每个WebSocketSession分配独立的锁对象，确保写操作同步
+ /** 会话锁映射：WebSocketSession -> 独立锁对象（防止并发写冲突） */
private final Map<WebSocketSession, Object> sessionLocks = new ConcurrentHashMap<>();
- // 存储用户最后心跳时间：userId -> 最后心跳时间戳（毫秒）
+ /** 用户最后心跳时间：userId -> 最后心跳时间戳（毫秒） */
private final Map<String, Long> userLastHeartbeatTime = new ConcurrentHashMap<>();
// ====================== 线程池 ======================
- // 延时任务线程池（单线程，守护线程），用于处理连接关闭后的延时清理
-// private final ScheduledExecutorService delayExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
-// Thread thread = new Thread(runnable, "websocket-关闭-延迟-线程");
-// thread.setDaemon(true); // 设为守护线程，不阻塞JVM退出
-// return thread;
-// });
-
- // 心跳检测线程池（单线程，守护线程），定时检查超时连接
+ /** 心跳检测线程池（单线程守护线程） */
private final ScheduledExecutorService heartbeatCheckExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
- Thread thread = new Thread(runnable, "websocket-心跳检测-线程");
- thread.setDaemon(true);
+ Thread thread = new Thread(runnable, "websocket-heartbeat-check-thread");
+ thread.setDaemon(true); // 守护线程，不阻塞JVM退出
return thread;
});
- // ====================== 初始化心跳检测任务 ======================
+ // ====================== 初始化 ======================
public DockWebSocketHandler() {
- // 项目启动后，每隔 CHECK_INTERVAL 秒执行一次心跳检测
- heartbeatCheckExecutor.scheduleAtFixedRate(this::checkHeartbeatTimeout,
- CHECK_INTERVAL, CHECK_INTERVAL, TimeUnit.SECONDS);
+ // 启动心跳检测定时任务：首次延迟CHECK_INTERVAL执行，之后每隔CHECK_INTERVAL执行
+ heartbeatCheckExecutor.scheduleAtFixedRate(
+ this::checkHeartbeatTimeout,
+ HEARTBEAT_CHECK_INTERVAL,
+ HEARTBEAT_CHECK_INTERVAL,
+ TimeUnit.SECONDS
+ );
+ log.info("WebSocket心跳检测任务已启动，检测间隔：{}秒，超时阈值：{}秒",
+ HEARTBEAT_CHECK_INTERVAL, HEARTBEAT_TIMEOUT);
}
+ // ====================== 连接生命周期 ======================
/**
- * 连接建立后回调
+ * 连接建立成功回调
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
- // 从会话属性中获取认证后的用户ID
+ // 从会话属性获取认证后的用户ID（需前置拦截器完成认证）
String userId = (String) session.getAttributes().get("userId");
if (userId == null) {
- session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证"));
+ session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证，拒绝连接"));
+ log.warn("WebSocket连接建立失败：会话无用户认证信息");
return;
}
+ // 初始化会话相关存储
userSessions.put(userId, session);
- // 为新会话初始化锁对象
sessionLocks.put(session, new Object());
- // 初始化用户最后心跳时间（连接建立即视为一次心跳）
userLastHeartbeatTime.put(userId, System.currentTimeMillis());
- log.info("用户[{}]已连接，当前在线人数：{}", userId, userSessions.size());
+ log.info("用户[{}]WebSocket连接建立成功，当前在线人数：{}", userId, userSessions.size());
}
/**
- * 处理客户端发送的消息（包含心跳消息）
+ * 连接关闭回调（同步清理，无延时）
+ */
+ @Override
+ public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+ String userId = (String) session.getAttributes().get("userId");
+ if (userId != null) {
+ // 同步清理所有关联数据
+ userSessions.remove(userId);
+ sessionLocks.remove(session);
+ userLastHeartbeatTime.remove(userId);
+ removeUserAllSubscriptions(userId);
+
+ log.info("用户[{}]WebSocket连接关闭（状态码：{}，原因：{}），当前在线人数：{}",
+ userId, status.getCode(), status.getReason(), userSessions.size());
+ } else {
+ // 无认证用户直接清理锁
+ sessionLocks.remove(session);
+ log.warn("无认证用户WebSocket连接关闭，状态码：{}", status.getCode());
+ }
+ }
+
+ // ====================== 消息处理 ======================
+ /**
+ * 处理客户端文本消息（包含心跳、订阅/取消订阅）
*/
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
String userId = (String) session.getAttributes().get("userId");
- String jsonMessage = message.getPayload();
+ if (userId == null) {
+ sendMessage(session, "未认证，无法处理消息");
+ return;
+ }
+ String jsonPayload = message.getPayload();
try {
- // 解析客户端消息
- WebSocketMessage webSocketMessage = JSON.parseObject(jsonMessage, WebSocketMessage.class);
- String type = webSocketMessage.getType();
+ // 解析WebSocket消息体
+ WebSocketMessage webSocketMessage = JSON.parseObject(jsonPayload, WebSocketMessage.class);
+ String messageType = webSocketMessage.getType();
Payload payload = webSocketMessage.getPayload();
- // ====================== 处理心跳消息 ======================
- if (MessageType.PING.getType().equals(type)) {
- // 1. 更新用户最后心跳时间
- userLastHeartbeatTime.put(userId, System.currentTimeMillis());
- // 2. 回复pong消息
- WebSocketMessage pongMessage = new WebSocketMessage();
- pongMessage.setType(MessageType.PONG.getType());
- pongMessage.setPayload(new Payload()); // 空payload即可
- sendMessage(session, JSON.toJSONString(pongMessage));
- log.debug("用户[{}]心跳检测成功，已回复pong", userId);
- System.out.println("用户["+userId+"]心跳检测成功，已回复pong");
- return; // 心跳消息无需走后续订阅逻辑
- }
-
- // 验证消息类型是否合法
- if (!MessageType.fromString(type)) {
- sendMessage(session, "不支持的消息类型：" + type);
+ // 1. 处理心跳PING消息
+ if (MessageType.PING.getType().equals(messageType)) {
+ handleHeartbeatPing(session, userId);
return;
}
- // 处理订阅/取消订阅
- handleSubscription(userId, type, payload);
+ // 2. 验证消息类型合法性
+ if (!MessageType.fromString(messageType)) {
+ sendMessage(session, "不支持的消息类型：" + messageType);
+ log.warn("用户[{}]发送非法消息类型：{}", userId, messageType);
+ return;
+ }
+
+ // 3. 处理订阅/取消订阅逻辑
+ handleSubscription(userId, messageType, payload);
} catch (Exception e) {
- log.error("用户[{}]消息处理错误", userId, e);
- sendMessage(session, "消息格式错误");
+ log.error("用户[{}]消息处理异常", userId, e);
+ sendMessage(session, "消息格式错误或处理失败：" + e.getMessage());
}
+ }
+
+ /**
+ * 处理心跳PING消息
+ */
+ private void handleHeartbeatPing(WebSocketSession session, String userId) throws IOException {
+ // 更新最后心跳时间
+ userLastHeartbeatTime.put(userId, System.currentTimeMillis());
+ // 回复PONG消息
+ WebSocketMessage pongMessage = new WebSocketMessage();
+ pongMessage.setType(MessageType.PONG.getType());
+ pongMessage.setPayload(new Payload());
+ sendMessage(session, JSON.toJSONString(pongMessage));
+
+ log.debug("用户[{}]心跳PING接收成功，已回复PONG", userId);
}
/**
@@ -140,91 +179,67 @@
*/
private void handleSubscription(String userId, String type, Payload payload) {
String sn = payload.getSn();
- boolean subscribe = payload.isSubscribe();
- String subscriptionKey = getSubscriptionKey(sn, type); // 生成唯一订阅标识
+ boolean isSubscribe = payload.isSubscribe();
+ String subscriptionKey = getSubscriptionKey(sn, type);
- if (subscribe) {
- // 订阅操作
+ if (isSubscribe) {
+ // 订阅操作：不存在则创建，存在则添加用户
subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
- log.info("用户[{}]订阅了 设备[{}]的[{}]类型消息", userId, sn, type);
+ log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type);
+
+ // 直播类型订阅：触发开启直播
if (MessageType.LIVESTREAM.getType().equals(type)) {
- // 开启直播
- log.info("开启设备[{}]直播", sn);
_liveStreamService.StartLiveStreaming(sn);
+ log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
}
} else {
// 取消订阅操作
if (subscriptions.containsKey(subscriptionKey)) {
- subscriptions.get(subscriptionKey).remove(userId);
- if (subscriptions.get(subscriptionKey).isEmpty()) {
+ Set<String> subscriberIds = subscriptions.get(subscriptionKey);
+ subscriberIds.remove(userId);
+ log.info("用户[{}]取消订阅设备[{}]的[{}]类型消息", userId, sn, type);
+
+ // 订阅组为空时移除，并停止直播（直播类型）
+ if (subscriberIds.isEmpty()) {
subscriptions.remove(subscriptionKey);
- // 若取消的是直播订阅且无其他订阅者，停止直播
if (MessageType.LIVESTREAM.getType().equals(type)) {
- log.info("设备[{}]无订阅者，停止直播", sn);
_liveStreamService.StopLiveStreaming(sn);
+ log.info("设备[{}]直播流已停止（无订阅用户）", sn);
}
}
- log.info("用户[{}]取消订阅了 设备[{}]的[{}]类型消息", userId, sn, type);
}
}
}
+ // ====================== 心跳检测 ======================
/**
- * 连接关闭后回调
- */
- @Override
- public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
- String userId = (String) session.getAttributes().get("userId");
- if (userId != null) {
- log.info("用户[{}]连接关闭（状态码：{}），将延时5秒执行清理操作", userId, status.getCode());
-
- // 延时5秒执行清理逻辑
-// delayExecutor.schedule(() -> {
- try {
- // 清理核心存储
- userSessions.remove(userId);
- sessionLocks.remove(session);
- userLastHeartbeatTime.remove(userId); // 清理心跳时间
-
- // 清除该用户的所有订阅
- removeUserAllSubscriptions(userId);
-
- log.info("用户[{}]延时5秒清理完成，当前在线人数：{}", userId, userSessions.size());
- } catch (Exception e) {
- log.error("用户[{}]延时清理操作异常", userId, e);
- }
-// }, 5, TimeUnit.SECONDS); // 延时5秒执行
- } else {
- // 无认证用户，直接清理
- sessionLocks.remove(session);
- log.info("无认证用户连接关闭，直接清理会话锁");
- }
- }
-
- /**
- * 心跳超时检测：断开超过HEARTBEAT_TIMEOUT未发心跳的用户
+ * 检测心跳超时连接并主动关闭
*/
private void checkHeartbeatTimeout() {
long currentTime = System.currentTimeMillis();
- // 遍历所有在线用户，检查心跳超时
+ // 遍历所有用户心跳记录，检查超时
for (Map.Entry<String, Long> entry : userLastHeartbeatTime.entrySet()) {
String userId = entry.getKey();
- Long lastHeartbeat = entry.getValue();
- WebSocketSession session = userSessions.get(userId);
+ long lastHeartbeatTime = entry.getValue();
+ long timeoutMs = currentTime - lastHeartbeatTime;
- // 1. 计算超时时间差
- long timeoutDiff = currentTime - lastHeartbeat;
- if (timeoutDiff > HEARTBEAT_TIMEOUT * 1000) {
- log.warn("用户[{}]心跳超时（{}秒未发ping），将断开连接", userId, timeoutDiff / 1000);
+ // 判定超时：超过阈值则主动关闭连接
+ if (timeoutMs > HEARTBEAT_TIMEOUT * 1000) {
+ WebSocketSession session = userSessions.get(userId);
+ log.warn("用户[{}]心跳超时（{}秒未发送PING），将主动关闭连接",
+ userId, timeoutMs / 1000);
+
try {
- // 2. 主动关闭连接
+ // 主动关闭连接
if (session != null && session.isOpen()) {
- session.close(CloseStatus.SESSION_NOT_RELIABLE.withReason("心跳超时"));
+ session.close(CloseStatus.SESSION_NOT_RELIABLE.withReason(
+ String.format("心跳超时：%d秒未收到PING", timeoutMs / 1000)
+ ));
}
} catch (IOException e) {
- log.error("关闭用户[{}]超时连接失败", userId, e);
+ log.error("主动关闭用户[{}]超时连接失败", userId, e);
} finally {
- // 3. 清理该用户的所有数据
+ // 强制清理用户所有关联数据
userSessions.remove(userId);
sessionLocks.remove(session);
userLastHeartbeatTime.remove(userId);
@@ -234,69 +249,84 @@
}
}
+ // ====================== 消息推送接口 ======================
/**
- * 清除用户的所有订阅
- */
- private void removeUserAllSubscriptions(String userId) {
- // 遍历所有订阅，移除该用户并检查是否需要停止直播
- Iterator<Map.Entry<String, Set<String>>> iterator = subscriptions.entrySet().iterator();
- while (iterator.hasNext()) {
- Map.Entry<String, Set<String>> entry = iterator.next();
- Set<String> userIds = entry.getValue();
- if (userIds.contains(userId)) {
- userIds.remove(userId);
- // 若订阅组为空，移除并停止直播（如果是直播类型）
- if (userIds.isEmpty()) {
- String subscriptionKey = entry.getKey();
- if (subscriptionKey.contains(MessageType.LIVESTREAM.getType())) {
- String sn = subscriptionKey.split(":")[0];
- _liveStreamService.StopLiveStreaming(sn);
- log.info("设备[{}]无订阅者，停止直播", sn);
- }
- iterator.remove(); // 安全移除空订阅组
- }
- }
- }
- }
-
- /**
- * 向订阅了特定设备特定类型的用户发送消息
+ * 向订阅指定设备指定类型的用户推送消息
+ *
+ * @param sn 设备序列号
+ * @param type 消息类型（对应MessageType）
+ * @param content 消息内容（JSON字符串）
*/
public void sendMessageToSubscribers(String sn, String type, String content) {
String subscriptionKey = getSubscriptionKey(sn, type);
- Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, new HashSet<>());
+ Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet());
if (subscriberIds.isEmpty()) {
+ log.debug("设备[{}]的[{}]类型消息无订阅用户，无需推送", sn, type);
return;
}
- try {
- for (String userId : subscriberIds) {
- WebSocketSession session = userSessions.get(userId);
- if (session == null || !session.isOpen()) {
- userSessions.remove(userId);
- sessionLocks.remove(session);
- userLastHeartbeatTime.remove(userId); // 清理心跳时间
- continue;
- }
-
- // 同步发送消息，避免并发冲突
- Object lock = sessionLocks.get(session);
- if (lock == null) {
- continue;
- }
- synchronized (lock) {
- session.sendMessage(new TextMessage(content));
- log.debug("向用户[{}]发送 设备[{}]的[{}]类型消息:{}", userId, sn, type, content);
- }
+ // 遍历订阅用户推送消息
+ for (String userId : subscriberIds) {
+ WebSocketSession session = userSessions.get(userId);
+ if (session == null || !session.isOpen()) {
+ // 清理无效会话
+ userSessions.remove(userId);
+ sessionLocks.remove(session);
+ userLastHeartbeatTime.remove(userId);
+ log.warn("用户[{}]会话无效，已清理", userId);
+ continue;
}
- } catch (IOException e) {
- log.error("发送消息错误：{}", e.getMessage());
+
+ // 线程安全发送消息
+ try {
+ sendMessage(session, content);
+ log.debug("向用户[{}]推送设备[{}]的[{}]类型消息：{}", userId, sn, type, content);
+ } catch (IOException e) {
+ log.error("向用户[{}]推送消息失败", userId, e);
+ }
}
}
/**
- * 通用发送消息方法（封装同步逻辑）
+ * 新增：向所有在线客户端广播消息
+ *
+ * @param content 消息内容（JSON字符串）
+ */
+ public void broadcastMessageToAllClients(String content) {
+ if (userSessions.isEmpty()) {
+ log.debug("无在线客户端，无需广播消息");
+ return;
+ }
+
+ log.info("开始向所有在线客户端（{}人）广播消息", userSessions.size());
+ // 遍历所有在线会话发送消息
+ for (Map.Entry<String, WebSocketSession> entry : userSessions.entrySet()) {
+ String userId = entry.getKey();
+ WebSocketSession session = entry.getValue();
+
+ if (session == null || !session.isOpen()) {
+ // 清理无效会话
+ userSessions.remove(userId);
+ sessionLocks.remove(session);
+ userLastHeartbeatTime.remove(userId);
+ log.warn("用户[{}]会话无效，已清理", userId);
+ continue;
+ }
+
+ // 线程安全发送广播消息
+ try {
+ sendMessage(session, content);
+ log.debug("向用户[{}]广播消息成功", userId);
+ } catch (IOException e) {
+ log.error("向用户[{}]广播消息失败", userId, e);
+ }
+ }
+ }
+
+ // ====================== 通用工具方法 ======================
+ /**
+ * 线程安全发送消息（封装同步逻辑）
*/
private void sendMessage(WebSocketSession session, String content) throws IOException {
Object lock = sessionLocks.get(session);
@@ -308,28 +338,61 @@
}
/**
- * 生成订阅标识（sn+type的组合）
+ * 生成订阅标识（设备SN + 消息类型）
*/
private String getSubscriptionKey(String sn, String type) {
- return sn + ":" + type;
+ return String.format("%s:%s", sn, type);
}
- // ====================== 销毁线程池（可选，防止内存泄漏） ======================
+ /**
+ * 清除用户所有订阅关系
+ */
+ private void removeUserAllSubscriptions(String userId) {
+ // 迭代器遍历，避免并发修改异常
+ Iterator<Map.Entry<String, Set<String>>> iterator = subscriptions.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<String, Set<String>> entry = iterator.next();
+ Set<String> subscriberIds = entry.getValue();
-// public void destroy() {
-// delayExecutor.shutdown();
-// heartbeatCheckExecutor.shutdown();
-// try {
-// if (!delayExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
-// delayExecutor.shutdownNow();
-// }
-// if (!heartbeatCheckExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
-// heartbeatCheckExecutor.shutdownNow();
-// }
-// } catch (InterruptedException e) {
-// delayExecutor.shutdownNow();
-// heartbeatCheckExecutor.shutdownNow();
-// }
-// log.info("WebSocket线程池已销毁");
-// }
+ // 移除当前用户
+ if (subscriberIds.remove(userId)) {
+ log.debug("用户[{}]已移除订阅：{}", userId, entry.getKey());
+
+ // 订阅组为空时移除，并停止直播（直播类型）
+ if (subscriberIds.isEmpty()) {
+ String subscriptionKey = entry.getKey();
+ if (subscriptionKey.contains(MessageType.LIVESTREAM.getType())) {
+ String sn = subscriptionKey.split(":")[0];
+ _liveStreamService.StopLiveStreaming(sn);
+ log.info("设备[{}]无订阅用户，已停止直播流", sn);
+ }
+ iterator.remove(); // 安全移除空订阅组
+ }
+ }
+ }
+ }
+
+ // ====================== 资源销毁 ======================
+ /**
+ * 容器销毁时关闭线程池，防止内存泄漏
+ */
+ @PreDestroy
+ public void destroy() {
+ // 关闭心跳检测线程池
+ if (heartbeatCheckExecutor != null && !heartbeatCheckExecutor.isShutdown()) {
+ heartbeatCheckExecutor.shutdown();
+ try {
+ // 等待1秒优雅关闭，超时则强制关闭
+ if (!heartbeatCheckExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+ heartbeatCheckExecutor.shutdownNow();
+ log.warn("WebSocket心跳检测线程池强制关闭");
+ }
+ } catch (InterruptedException e) {
+ heartbeatCheckExecutor.shutdownNow();
+ Thread.currentThread().interrupt(); // 恢复中断状态
+ log.error("WebSocket心跳检测线程池关闭被中断", e);
+ }
+ }
+ log.info("WebSocket处理器资源已销毁");
+ }
}
\ No newline at end of file
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/model/dto/DeviceDTO.java b/ard-sample/src/main/java/com/ard/sample/manage/model/dto/DeviceDTO.java
index e12138e..1c4ef37 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/model/dto/DeviceDTO.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/model/dto/DeviceDTO.java
@@ -70,4 +70,6 @@
private String webrtcUrl;
private String cameraWebrtcUrl;
+
+ private String cameraState;
}
\ No newline at end of file
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/model/entity/DeviceEntity.java b/ard-sample/src/main/java/com/ard/sample/manage/model/entity/DeviceEntity.java
index 6bb7b84..9fa1f67 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/model/entity/DeviceEntity.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/model/entity/DeviceEntity.java
@@ -100,4 +100,7 @@
@TableField(value = "cameraWebrtcUrl")
private String cameraWebrtcUrl;
+ @TableField(value = "cameraState")
+ private String cameraState;
+
}
\ No newline at end of file
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/LiveStatusTwoReceiver.java b/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/LiveStatusTwoReceiver.java
new file mode 100644
index 0000000..79b9166
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/LiveStatusTwoReceiver.java
@@ -0,0 +1,37 @@
+package com.ard.sample.manage.model.receiver;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+@Data
+public class LiveStatusTwoReceiver {
+ /**
+ * 直播码流标识符
+ * */
+ @JsonProperty("video_id")
+ private String videoId;
+
+ /**
+ * 视频类型
+ * */
+ @JsonProperty("video_type")
+ private String videoType;
+
+ /**
+ * 直播码流的质量 "0":"自适应","1":"流畅","2":"标清","3":"高清","4":"超清"
+ * */
+ @JsonProperty("video_quality")
+ private Integer videoQuality;
+
+ /**
+ * 直播状态 "0":"未直播","1":"在直播"
+ * */
+ @JsonProperty("status")
+ private Integer status;
+
+ /**
+ * 错误码
+ * */
+ @JsonProperty("error_status")
+ private Integer errorStatus;
+}
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/OsdGatewayReceiver.java b/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/OsdGatewayReceiver.java
index 08066eb..c4cc145 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/OsdGatewayReceiver.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/model/receiver/OsdGatewayReceiver.java
@@ -25,6 +25,7 @@
private Integer transmissionSignalQuality;
+ @JsonProperty("live_status")
private LiveStatusReceiver liveStatus;
private WirelessLinkStateReceiver wirelessLinkState;
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/IDeviceService.java b/ard-sample/src/main/java/com/ard/sample/manage/service/IDeviceService.java
index 1f3d583..e7301e1 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/IDeviceService.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/IDeviceService.java
@@ -263,6 +263,9 @@
*/
Boolean checkDeviceOnline(String sn);
+ void sendCameraInfo();
+ void setRedisCameraPosition(String sn, Integer cameraPosition);
+ void setChildRedisCamera(String sn, String context);
// public List<DeviceEntity> selectList(Wrapper<DeviceEntity> queryWrapper);
}
\ No newline at end of file
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java
index 8e315ce..0645288 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java
@@ -63,6 +63,7 @@
return ResponseResult.error("The " + serviceIdentifier + " method does not exist.");
}
+
Object data = "";
// Add parameter validation.
if (Objects.nonNull(controlMethodEnum.getClazz())) {
@@ -111,6 +112,18 @@
RedisOpsUtils.setWithExpire(serviceIdentifier + RedisConst.DELIMITER + bid, sn,
RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND);
}
+ // 执行操作成功后，更新摄像头内外镜头状态
+ if(RemoteControlMethodEnum.COVER_OPEN == controlMethodEnum) {
+ // 打开盖子,切换机场舱内摄像头镜头
+ //打开，摄像头为舱内；将摄像头状态存到redis，格式为：CONVERSTATE:设备SN：1（舱外）或0（舱内）
+ _deviceService.setRedisCameraPosition(sn,0);
+ }
+ if(RemoteControlMethodEnum.COVER_CLOSE == controlMethodEnum) {
+ // 关闭盖子,切换机场舱外摄像头镜头
+ //关闭，摄像头为舱外；将摄像头状态存到redis，格式为：CONVERSTATE:设备SN：1（舱外）或0（舱内）
+ _deviceService.setRedisCameraPosition(sn,1);
+ }
+
return ResponseResult.success();
}
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java
index c66e278..7683a49 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java
@@ -4,6 +4,8 @@
import com.ard.device.camera.domain.ArdCamera;
import com.ard.device.camera.service.IArdCameraService;
import com.ard.device.channel.domain.ArdChannel;
+import com.ard.sample.component.websocket.enums.MessageType;
+import com.ard.sample.component.websocket.handler.DockWebSocketHandler;
import com.ard.sample.component.websocket.handler.LiveStreamHandler;
import com.ard.system.authorize.service.AuthorizeService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
@@ -38,12 +40,14 @@
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
+import org.apache.poi.ss.formula.functions.T;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
+import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
@@ -56,6 +60,7 @@
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static com.ard.sample.component.mqtt.model.TopicConst.*;
@@ -69,6 +74,9 @@
@Slf4j
@Transactional
public class DeviceServiceImpl implements IDeviceService {
+
+ @Autowired
+ private DockWebSocketHandler _dockWebSocketHandler;
@Autowired
private LiveStreamHandler _liveStreamHandler;
@@ -377,7 +385,7 @@
for(ArdChannel ardChannel : ardCamera1.getChannelList()) {
if(ardChannel.getChanNo() == 1) {
// 修改列表中外接相机地址
- updateCameraWebrtcUrl(devices.getDeviceSn(), ardChannel.getArdVtdu().getWebrtcUrl());
+ updateCameraWebrtcUrl(devices.getDeviceSn(), ardChannel.getArdVtdu().getWebrtcUrl(), ardCamera1.getState());
}
}
}
@@ -867,6 +875,7 @@
.workType(entity.getWorkType())
.webrtcUrl(entity.getWebrtcUrl())
.cameraWebrtcUrl(entity.getCameraWebrtcUrl())
+ .cameraState(entity.getCameraState())
.workspaceName(entity.getDeptId() != null
? Optional.ofNullable(sysDeptService.selectDeptById(entity.getDeptId()))
.map(SysDept::getDeptName)
@@ -927,10 +936,12 @@
return update > 0;
}
- public Boolean updateCameraWebrtcUrl(String sn, String CameraWebrtcUrl){
+ public Boolean updateCameraWebrtcUrl(String sn, String CameraWebrtcUrl, String state){
LambdaUpdateWrapper<DeviceEntity> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
lambdaUpdateWrapper.eq(DeviceEntity::getDeviceSn, sn)
- .set(DeviceEntity::getCameraWebrtcUrl, CameraWebrtcUrl);
+ .set(DeviceEntity::getCameraWebrtcUrl, CameraWebrtcUrl)
+ .set(DeviceEntity::getCameraState, state);
+
int update = mapper.update(null, lambdaUpdateWrapper);
return update > 0;
}
@@ -1328,6 +1339,7 @@
.workType(dto.getWorkType())
.webrtcUrl(dto.getWebrtcUrl())
.cameraWebrtcUrl(dto.getCameraWebrtcUrl())
+ .cameraState(dto.getCameraState())
.compatibleStatus(dto.getFirmwareStatus() == null ? null :
DeviceFirmwareStatusEnum.CONSISTENT_UPGRADE != DeviceFirmwareStatusEnum.find(dto.getFirmwareStatus()))
.build();
@@ -1469,9 +1481,48 @@
messageSender.publish(topic, builder.build());
}
-// @DataScopePlus(deptAlias="d")
-// @Override
-// public List<DeviceEntity> selectList(Wrapper<DeviceEntity> queryWrapper) {
-// return mapper.selectList(queryWrapper);
-// }
+ /**
+ * 定时任务，每秒钟从redis中读取信息发送机场摄像头（舱内舱外）、飞行器摄像头模式（广角、变焦、红外）
+ * */
+ @Override
+ @Scheduled(initialDelay = 5, fixedRate = 1, timeUnit = TimeUnit.SECONDS)
+ public void sendCameraInfo() {
+ // 此方法修改为定时任务，每秒钟从redis中读取信息发送
+ // 发送机场摄像头位置信息
+ Map<Object, Object> map = RedisOpsUtils.getAllData(RedisConst.COVER_STATE_PREFIX+"*");
+ map.forEach((k, v) -> {
+ _dockWebSocketHandler.broadcastMessageToAllClients(v.toString());
+ });
+ }
+
+ // 机场直播摄像头信息存redis中，舱内、舱外
+ @Override
+ public void setRedisCameraPosition(String sn, Integer cameraPosition) {
+ String liveStatus = "";
+ Optional<DeviceDTO> device = getDeviceBySn(sn);
+ if(device.get().getDomain().equals(DeviceDomainEnum.DOCK.getVal())){
+ CommonTopicReceiver<Map<String,Integer>> stateReceiver = new CommonTopicReceiver<>();
+ stateReceiver.setTid(UUID.randomUUID().toString());
+ stateReceiver.setBid(UUID.randomUUID().toString());
+ stateReceiver.setType(MessageType.LIVESTREAM.getType());
+ stateReceiver.setGateway(sn);
+ stateReceiver.setTimestamp(System.currentTimeMillis());
+ Map<String,Integer> live_status = new HashMap<>();
+ live_status.put("camera_position", cameraPosition);
+ stateReceiver.setData(live_status);
+ try{
+ liveStatus=objectMapper.writeValueAsString(stateReceiver);
+ }
+ catch (Exception e){
+ log.error("机场直播设备状态反馈转换异常:{}",e);
+ }
+ }
+ log.info("机场直播设备状态反馈存入redis中，key值：{}中:{}",RedisConst.COVER_STATE_PREFIX,liveStatus);
+ RedisOpsUtils.set(RedisConst.COVER_STATE_PREFIX + sn, liveStatus);
+ }
+
+ public void setChildRedisCamera(String sn, String context){
+ // 保存子设备信息到redis中，便于前端获取
+ RedisOpsUtils.set(RedisConst.COVER_STATE_PREFIX + sn, context);
+ }
}
\ No newline at end of file
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java
index 8bdc47d..6e5806c 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java
@@ -51,10 +51,12 @@
_deviceService.updateWorkType(device.getDeviceSn(), 3L);
RedisOpsUtils.setWithExpire(RedisConst.OSD_STATUS_PREFIX + device.getDeviceSn(), 3, RedisConst.DEVICE_ALIVE_SECOND);
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getDeviceSn());
+ RedisOpsUtils.del(RedisConst.COVER_STATE_PREFIX + device.getDeviceSn());
}
if(!_deviceService.checkDeviceOnline(device.getChildDeviceSn())){
_deviceService.updateWorkType(device.getChildDeviceSn(), 6L);
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getChildDeviceSn());
+ RedisOpsUtils.del(RedisConst.COVER_STATE_PREFIX + device.getDeviceSn());
}
}
}
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/LiveStreamServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/LiveStreamServiceImpl.java
index e612f7b..d7bbe16 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/LiveStreamServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/LiveStreamServiceImpl.java
@@ -182,6 +182,11 @@
return ResponseResult.error(LiveErrorEnum.URL_TYPE_NOT_SUPPORTED);
}
+ // 开始直播时发送机场相机摄像头默认信息（舱外）到所有在线用户,只取机场sn，发机场的摄像头信息
+ if(data.getChildDeviceSn()!=null) {
+ String sn = data.getDeviceSn();
+ deviceService.setRedisCameraPosition(sn, 1);
+ }
return ResponseResult.success(live);
}
@@ -268,6 +273,10 @@
if (ResponseResult.CODE_SUCCESS != receiveReply.getResult()) {
return ResponseResult.error(LiveErrorEnum.find(receiveReply.getResult()));
}
+ // 将切换相机摄像头信息到redis
+ Integer camera_position = liveParam.getCameraPosition();
+ String sn = responseResult.getData().getDeviceSn();
+ deviceService.setRedisCameraPosition(sn, camera_position);
return ResponseResult.success();
}
--
Gitblit v1.9.3