From 0ab71d4f9154dc2a0306ea57ec08659f4e28e449 Mon Sep 17 00:00:00 2001
From: Xiajijian <1176061145@qq.com>
Date: Mon, 16 Mar 2026 16:04:06 +0800
Subject: [PATCH] 1. 智云3.0业务优化，优化类型获取。
---
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 33 +++++++++++++++++++++------------
1 files changed, 21 insertions(+), 12 deletions(-)
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 51c8a5a..2d39797 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
@@ -4,7 +4,6 @@
import com.ard.sample.component.websocket.enums.MessageType;
import com.ard.sample.component.websocket.model.Payload;
import com.ard.sample.component.websocket.model.WebSocketMessage;
-import com.ard.sample.manage.service.IControlService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -31,9 +30,6 @@
@Autowired
private LiveStreamHandler _liveStreamService;
-
- @Autowired
- private IControlService _controlService;
// ====================== 心跳相关常量（可配置化建议放到配置文件） ======================
/** 心跳间隔（秒）：客户端需每隔此时间发送一次PING消息 */
@@ -153,6 +149,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());
@@ -185,13 +188,13 @@
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)) {
- _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 +260,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 +283,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) {
@@ -299,7 +308,7 @@
return;
}
- log.info("开始向所有在线客户端（{}人）广播消息", userSessions.size());
+ // log.info("开始向所有在线客户端（{}人）广播消息", userSessions.size());
// 遍历所有在线会话发送消息
for (Map.Entry<String, WebSocketSession> entry : userSessions.entrySet()) {
String userId = entry.getKey();
--
Gitblit v1.9.3