forked from uav/ard-uav-server
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; 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) 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; } } 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); ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java
New file @@ -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; } } 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);
