From 7beea5916728a84617d4791775d0aac70ab32eee Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Fri, 12 Dec 2025 10:40:43 +0800
Subject: [PATCH] 2025-12-12 zwy 优化，订阅视频时，谁订阅发给谁直播地址，其他订阅不变
---
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceOSDServiceImpl.java | 2 +-
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java | 2 +-
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 25 +++++++++++++++++++------
ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java | 2 +-
ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/PropertySetReplyHandler.java | 2 +-
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java | 2 +-
ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java | 4 ++--
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java | 7 +++++--
ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java | 2 +-
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java | 4 ++--
10 files changed, 34 insertions(+), 18 deletions(-)
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/PropertySetReplyHandler.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/PropertySetReplyHandler.java
index 4bb5a40..f84e890 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/PropertySetReplyHandler.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/PropertySetReplyHandler.java
@@ -60,6 +60,6 @@
String dockControl="";
dockControl=mapper.writeValueAsString(receiver);
- _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl);
+ _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl, "");
}
}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java
index c1909e5..c7d6e2f 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java
@@ -64,7 +64,7 @@
String dockControl="";
dockControl=mapper.writeValueAsString(receiver);
- _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl);
+ _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl, "");
// DockControlProgressWebSockerHandler.
}
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 c9af46c..36ab8cc 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
@@ -153,6 +153,13 @@
// 3. 处理订阅/取消订阅逻辑
handleSubscription(userId, messageType, payload);
+ // 直播类型订阅：触发开启直播
+ if (MessageType.LIVESTREAM.getType().equals(messageType) && payload.isSubscribe()) {
+ String sn = payload.getSn();
+ _liveStreamService.StartLiveStreaming(sn,userId);
+ log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
+ }
+
} catch (Exception e) {
log.error("用户[{}]消息处理异常", userId, e);
sendMessage(session, "消息格式错误或处理失败：" + e.getMessage());
@@ -187,11 +194,11 @@
subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
// log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type);
- // 直播类型订阅：触发开启直播
- if (MessageType.LIVESTREAM.getType().equals(type)) {
- _liveStreamService.StartLiveStreaming(sn);
- log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
- }
+// // 直播类型订阅：触发开启直播
+// if (MessageType.LIVESTREAM.getType().equals(type)) {
+// _liveStreamService.StartLiveStreaming(sn);
+// log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
+// }
} else {
// 取消订阅操作
if (subscriptions.containsKey(subscriptionKey)) {
@@ -257,7 +264,7 @@
* @param type 消息类型（对应MessageType）
* @param content 消息内容（JSON字符串）
*/
- public void sendMessageToSubscribers(String sn, String type, String content) {
+ public void sendMessageToSubscribers(String sn, String type, String content, String uId) {
String subscriptionKey = getSubscriptionKey(sn, type);
Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet());
@@ -280,6 +287,12 @@
// 线程安全发送消息
try {
+ if(MessageType.LIVESTREAM.getType().equals(type)) {
+ WebSocketSession liveSession = userSessions.get(uId);
+ sendMessage(liveSession, content);
+ log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息：{}", uId, sn, type, content);
+ return;
+ }
sendMessage(session, content);
log.debug("向用户[{}]推送设备[{}]的[{}]类型消息：{}", userId, sn, type, content);
} catch (IOException e) {
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java
index 90e1f72..60c1ef7 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java
@@ -31,7 +31,7 @@
/**
* 开启直播
*/
- public void StartLiveStreaming(String sn) {
+ public void StartLiveStreaming(String sn ,String userId) {
//调用开始直播
ResponseResult rlive = null;
String resultLive = "";
@@ -61,7 +61,7 @@
}
}
}
- _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), resultLive);
+ _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), resultLive, userId);
log.info("SN: {}-直播地址：{}", sn, resultLive);
} catch (Exception e) {
log.info("SN: {}-开启直播失败！原因：{}", sn, e.getMessage());
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 b2c1dbd..ca96e58 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
@@ -191,7 +191,7 @@
e.printStackTrace();
}
// 设备控制进度消息websocket发给前端
- _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.PROGRESS.getType(), dockProgress);
+ _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.PROGRESS.getType(), dockProgress, "");
if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) {
String topic = headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF;
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceOSDServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceOSDServiceImpl.java
index 9f55a60..b9c3894 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceOSDServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceOSDServiceImpl.java
@@ -116,7 +116,7 @@
wsMessage.getData().setHost(data);
// 发送给websocket客户端
- _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd);
+ _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd, "");
sendMessageService.sendBatch(webSessions, wsMessage);
this.pushTelemetryData(data, device.getDeviceSn());
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 9b8d260..efc6fba 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
@@ -1,6 +1,8 @@
package com.ard.sample.manage.service.impl;
import com.alibaba.fastjson2.JSONObject;
+import com.ard.common.core.domain.model.LoginUser;
+import com.ard.common.utils.SecurityUtils;
import com.ard.device.camera.domain.ArdCamera;
import com.ard.device.camera.service.IArdCameraService;
import com.ard.device.channel.domain.ArdChannel;
@@ -369,6 +371,7 @@
//region 单独写一个方法做权限验证，由于有好多定时任务也在实时获取设备列表，原方法不适用再添加权限注解
@Override
public List<DeviceDTO> getDockList(String nickname) {
+ LoginUser loginUser = SecurityUtils.getLoginUser();
List<DeviceDTO> devicesList = this.getDockListByParams(
DeviceQueryParam.builder()
.domains(List.of(DeviceDomainEnum.GATEWAY.getVal(), DeviceDomainEnum.DOCK.getVal()))
@@ -376,8 +379,8 @@
.build());
// 在獲取设备列表时，顺便获取直播地址
for(DeviceDTO devices : devicesList){
- _liveStreamHandler.StartLiveStreaming(devices.getDeviceSn());
- _liveStreamHandler.StartLiveStreaming(devices.getChildDeviceSn());
+ _liveStreamHandler.StartLiveStreaming(devices.getDeviceSn(), loginUser.getUserId().toString());
+ _liveStreamHandler.StartLiveStreaming(devices.getChildDeviceSn(), loginUser.getUserId().toString());
ArdCamera ardCamera = new ArdCamera();
ardCamera.setDockId(devices.getId());
List<ArdCamera> ardCameraList = _cameraService.selectArdCameraList(ardCamera);
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java
index 18a8e27..36346d8 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java
@@ -97,7 +97,7 @@
wsMessage.getData().setHost(data);
sendMessageService.sendBatch(webSessions, wsMessage);
// 发送给websocket客户端
- _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd);
+ _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd, "");
}
}
}
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 670e1c2..9824635 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
@@ -84,7 +84,7 @@
String json = getJSONString(MessageType.STATE.getType(), device.getDeviceSn(), dockStatus, droneStatus);
// 发送给websocket客户端
- _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATE.getType(), json);
+ _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATE.getType(), json, "");
}
}
diff --git a/ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
index 34da626..0907f99 100644
--- a/ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java
@@ -149,7 +149,7 @@
} catch (Exception e) {
e.printStackTrace();
}
- _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json);
+ _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json, "");
}
/**
@@ -194,7 +194,7 @@
} catch (Exception e) {
e.printStackTrace();
}
- _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json);
+ _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json, "");
log.info("定时任务自定义状态为待执行，json: {}", json);
// endregion
long now = System.currentTimeMillis();
--
Gitblit v1.9.3