From d68181ce2faef4821d6d8a8ffb8027480a6262c3 Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Wed, 05 Nov 2025 17:01:20 +0800
Subject: [PATCH] 2025-11-5 zwy websockeet增加Type:livestream,开启机场和飞行器的直播接口， 增加LiveStreamHandler对开启直播和关闭直播的处理； redis增加WEBSOCKET_VideoId_PREFIX，存储videoid，关闭直播功能时使用；
---
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 21 ++++++
ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java | 14 +++-
ard-sample/src/main/java/com/ard/sample/common/model/ResponseResult.java | 2
ard-common/src/main/java/com/ard/common/core/redis/RedisConst.java | 2
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java | 2
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java | 138 ++++++++++++++++++++++++++++++++++++++++++++++
6 files changed, 173 insertions(+), 6 deletions(-)
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 f3c7ec0..b408979 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
@@ -19,6 +19,8 @@
public static final Integer WEBSOCKET_ALIVE_SECOND = 60 * 60 * 24;
+ public static final String WEBSOCKET_VideoId_PREFIX = "videoid" + 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/common/model/ResponseResult.java b/ard-sample/src/main/java/com/ard/sample/common/model/ResponseResult.java
index 50c2c94..809b7a1 100644
--- a/ard-sample/src/main/java/com/ard/sample/common/model/ResponseResult.java
+++ b/ard-sample/src/main/java/com/ard/sample/common/model/ResponseResult.java
@@ -24,6 +24,8 @@
private T data;
+ private String gateway;
+
public static <T> ResponseResult<T> success(T data) {
return ResponseResult.<T>builder()
.code(CODE_SUCCESS)
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 e46ea77..631dbe9 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
@@ -5,6 +5,7 @@
STATUS("status"),
CONTROL("control"),
PROGRESS("progress"),
+ LIVESTREAM("livestream"),
ALL("all");
private final String type;
@@ -17,12 +18,19 @@
return type;
}
- public static MessageType fromString(String type) {
+ public static boolean fromString(String type) {
+ // 遍历MessageType枚举的所有值
for (MessageType messageType : MessageType.values()) {
+ // 如果当前枚举值的type属性与传入的type字符串忽略大小写相等
if (messageType.type.equalsIgnoreCase(type)) {
- return messageType;
+ // 返回当前枚举值
+ //return messageType;
+ return true;
}
}
- throw new IllegalArgumentException("Unknown message type: " + type);
+ // 如果遍历完所有枚举值都没有找到匹配的，则抛出异常
+ //throw new IllegalArgumentException("Unknown message type: " + type);
+ return false;
}
+
}
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 89b92a5..9d6cc61 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,8 +1,11 @@
package com.ard.sample.component.websocket.handler;
import com.alibaba.fastjson2.JSON;
+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 org.checkerframework.checker.units.qual.A;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
@@ -15,6 +18,9 @@
@Component
public class DockWebSocketHandler extends TextWebSocketHandler {
+
+ @Autowired
+ private LiveStreamHandler _liveStreamService;
// 存储用户会话：userId -> WebSocketSession
private final Map<String, WebSocketSession> userSessions = new ConcurrentHashMap<>();
@@ -58,11 +64,11 @@
Payload payload = webSocketMessage.getPayload();
// 验证消息类型是否合法
- if (!Arrays.asList("status", "control", "progress", "state").contains(type)) {
+ //if (!Arrays.asList("status", "control", "progress", "state").contains(type)) {
+ if(!MessageType.fromString(type)){
session.sendMessage(new TextMessage("不支持的消息类型：" + type));
return;
}
-
// 处理订阅/取消订阅
handleSubscription(userId, type, payload);
@@ -84,7 +90,18 @@
// 订阅操作
subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
System.out.println("用户[" + userId + "]订阅了 设备[" + sn + "]的[" + type + "]类型消息");
+ if(MessageType.LIVESTREAM.getType().equals(type)){
+ // 开启直播
+ // 开启直播逻辑
+ System.out.println("开启直播");
+ _liveStreamService.StartLiveStreaming(sn);
+ }
} else {
+ // 停止直播
+ if(MessageType.LIVESTREAM.getType().equals(type)){
+ System.out.println("停止直播");
+ _liveStreamService.StopLiveStreaming(sn);
+ }
// 取消订阅操作
if (subscriptions.containsKey(subscriptionKey)) {
subscriptions.get(subscriptionKey).remove(userId);
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
new file mode 100644
index 0000000..1ae5cec
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java
@@ -0,0 +1,138 @@
+package com.ard.sample.component.websocket.handler;
+
+import com.ard.common.core.redis.RedisConst;
+import com.ard.common.core.redis.RedisOpsUtils;
+import com.ard.sample.common.model.ResponseResult;
+import com.ard.sample.component.websocket.enums.MessageType;
+import com.ard.sample.manage.model.dto.*;
+import com.ard.sample.manage.service.IDeviceService;
+import com.ard.sample.manage.service.ILiveStreamService;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+@Component
+public class LiveStreamHandler {
+ @Autowired
+ private IDeviceService _deviceService;
+
+ @Autowired
+ private ILiveStreamService _liveStreamService;
+
+ @Autowired
+ private DockWebSocketHandler _dockWebSocketHandler;
+
+ @Autowired
+ protected ObjectMapper _objectMapper;
+
+ /**
+ * 开启直播
+ */
+ public void StartLiveStreaming(String sn) {
+ // 检查设备是否在线
+ if (!_deviceService.checkDeviceOnline(sn)) {
+ String jsonString = getResult(sn, "设备不在线，请检查网络连接！");
+ _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), jsonString);
+ return;
+ }
+ // 读取直播能力拼接参数
+ List<CapacityDeviceDTO> devicesList = _liveStreamService.getLiveCapacity();
+ if (devicesList == null || devicesList.isEmpty()) {
+ String jsonString = getResult(sn, "未获得设备直播能力！");
+ _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), jsonString);
+ return;
+ }
+ for (CapacityDeviceDTO device : devicesList) {
+ if (!device.getSn().equals(sn)) {
+ continue;
+ }
+ List<CapacityCameraDTO> camerasList = device.getCamerasList();
+ for (CapacityCameraDTO camera : camerasList) {
+ if (camera.getName()==null || camera.getName().isEmpty()) {
+ continue;
+ }
+ LiveTypeDTO param = new LiveTypeDTO();
+ String videoid = device.getSn() + "/" + camera.getIndex();
+ List<CapacityVideoDTO> videosList = camera.getVideosList();
+ for (CapacityVideoDTO video : videosList) {
+ videoid = videoid + "/" + video.getIndex();
+ param.setUrlType(1);
+ param.setUrl("");
+ param.setVideoId(videoid);
+ param.setVideoQuality(1);
+ param.setVideoType(video.getSwitchVideoTypes().get(0));
+ //调用开始直播
+ ResponseResult rlive = null;
+ String resultLive = "";
+ try {
+ rlive = _liveStreamService.liveStart(param);
+ if (rlive != null) {
+ if (rlive.getCode() == 513003) {
+ _liveStreamService.liveStop(param.getVideoId());
+ System.out.println("相机已经在直播中，关闭直播后重启直播！");
+ rlive = _liveStreamService.liveStart(param);
+ }
+ if(rlive.getCode() == 0){
+ // 直播成功后，在redis中存一个标记，将videoid保留，留给停止直播使用
+ System.out.println("开启直播成功！");
+ RedisOpsUtils.setWithExpire(RedisConst.WEBSOCKET_VideoId_PREFIX + sn, param.getVideoId(), RedisConst.WEBSOCKET_ALIVE_SECOND);
+ rlive.setGateway(sn);
+ }
+ }
+ resultLive = _objectMapper.writeValueAsString(rlive);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), resultLive);
+ break;
+ }
+ }
+ }
+ }
+
+ public void StopLiveStreaming(String sn) {
+ // 停止直播逻辑
+ ResponseResult rlive = null;
+ String resultLive = "";
+// System.out.println("关闭直播");
+ String videoid = "";
+ if(!RedisOpsUtils.checkExist(RedisConst.WEBSOCKET_VideoId_PREFIX + sn)){
+ resultLive = getResult(sn, "未找到正在直播的相机！");
+ }
+ else {
+ videoid = String.valueOf(RedisOpsUtils.get(RedisConst.WEBSOCKET_VideoId_PREFIX + sn));
+ try {
+ rlive = _liveStreamService.liveStop(videoid);
+ rlive.setGateway(sn);
+ resultLive = _objectMapper.writeValueAsString(rlive);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), resultLive);
+ }
+
+ public String getResult(String sn, String msg) {
+ Map<String, String> result = new HashMap<>();
+ result.put("gateway", sn);
+ result.put("code", "1");
+ result.put("message", msg);
+ result.put("data", null);
+ result.put("type", MessageType.LIVESTREAM.getType());
+ // 将Map转换为JSON字符串
+ String jsonString = "";
+ try {
+ jsonString = _objectMapper.writeValueAsString(result);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ return jsonString;
+ }
+}
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 3d30a70..bab84be 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
@@ -61,7 +61,7 @@
dockStatus = String.valueOf(RedisOpsUtils.get(RedisConst.OSD_STATUS_PREFIX + device.getDeviceSn()));
}
if(RedisOpsUtils.checkExist(RedisConst.OSD_STATUS_PREFIX + device.getChildDeviceSn())){
- droneStatus = String.valueOf(RedisOpsUtils.get(RedisConst.OSD_STATUS_PREFIX + device.getChildDeviceSn()));;
+ droneStatus = String.valueOf(RedisOpsUtils.get(RedisConst.OSD_STATUS_PREFIX + device.getChildDeviceSn()));
}
String json = getJSONString(MessageType.STATE.getType(), device.getDeviceSn(), dockStatus, droneStatus);
--
Gitblit v1.9.3