forked from uav/ard-uav-server
ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
@@ -630,14 +630,14 @@ UavStateCmd uavStateCmd = (UavStateCmd) tyCommand; UavStateCache.putUavState(uavStateCmd.getUavState()); logPrint("无人机状态", uavStateCmd.getUavState().getBoxSn()); WebSocketUtil.sendMessageAll(uavStateCmd.getUavState()); // WebSocketUtil.sendMessageAll(uavStateCmd.getUavState()); break; case (byte) 0xA9: // 处理无人机遥测消息 UavTelemetryDataCmd uavTelemetryDataCmd = (UavTelemetryDataCmd) tyCommand; UavTelemetryDataCache.putUavTelemetryData(uavTelemetryDataCmd.getTelemetryData()); logPrint("无人机遥测", uavTelemetryDataCmd.getTelemetryData().getBoxSn()); WebSocketUtil.sendMessageAll(uavTelemetryDataCmd.getTelemetryData()); //WebSocketUtil.sendMessageAll(uavTelemetryDataCmd.getTelemetryData()); break; case (byte) 0x0C: ard-work/src/main/java/com/ard/tianyu/websocket/domain/UavStateMessage.java
@@ -27,7 +27,6 @@ */ private UavState uavState; /** * 是否订阅 */ ard-work/src/main/java/com/ard/tianyu/websocket/domain/UavStreamMessage.java
@@ -21,7 +21,7 @@ */ private String boxSn; /** * 类型 0-未拉流 1-拉流 * 是否订阅 false-未拉流 true-拉流 */ private Integer pullType; private Boolean subscribe; } ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
@@ -66,8 +66,8 @@ if (uavStreamMessage != null) { String boxSn = uavStreamMessage.getBoxSn(); // 根据pullType决定操作类型：1表示开始流，其他值表示停止流 if (uavStreamMessage.getPullType().equals(1)) { // 根据Subscribe决定操作类型：true表示开始拉流，false表示停止拉流 if (uavStreamMessage.getSubscribe()) { uavStreamManager.startPullStream(boxSn, userId); } else { uavStreamManager.stopPullStream(boxSn, userId); ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStateManager.java
@@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONObject; import com.ard.tianyu.domain.cache.UavStateCache; import com.ard.tianyu.domain.dto.state.UavState; import com.ard.tianyu.websocket.domain.BaseMessage; import com.ard.tianyu.websocket.utils.WebSocketUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -128,12 +129,13 @@ private void sendDroneStateToSubscribers(String boxSn) { // 获取当前无人机状态 UavState uavState = getDroneState(boxSn); BaseMessage<UavState>baseMessage=new BaseMessage<>("state",uavState); // 向所有订阅者推送状态 Set<Long> userIds = stateSubscriptions.get(boxSn); if (userIds != null) { for (Long userId : userIds) { try { WebSocketUtil.sendMessage(userId, JSON.toJSONString(uavState)); WebSocketUtil.sendMessage(userId, JSON.toJSONString(baseMessage)); } catch (Exception e) { log.error("Error sending state message to session {}", userId, e); } ard-work/src/main/java/com/ard/tianyu/websocket/handler/UavStreamManager.java
@@ -1,7 +1,10 @@ package com.ard.tianyu.websocket.handler; import com.alibaba.fastjson2.JSON; import com.ard.tianyu.domain.dto.state.UavState; import com.ard.tianyu.domain.receiver.BoxStreamUrlReceiver; import com.ard.tianyu.service.ITyService; import com.ard.tianyu.websocket.domain.BaseMessage; import com.ard.tianyu.websocket.utils.WebSocketUtil; import com.ard.vtdu.domain.ArdVtdu; import com.ard.vtdu.service.IMediaService; @@ -19,9 +22,9 @@ public class UavStreamManager { private static final String STREAM_LOG_PREFIX = "【stream】"; private static final String STREAM_NOT_FOUND_MSG = STREAM_LOG_PREFIX + "Box stream not found: "; private static final String STREAM_CLOSED_MSG = STREAM_LOG_PREFIX + "Stream closed due to no active users: "; private static final String STREAM_IN_USE_MSG = STREAM_LOG_PREFIX + "Stream is still in use, remaining users: "; private static final String STREAM_NOT_FOUND_MSG = "Box stream not found: "; private static final String STREAM_CLOSED_MSG = "Stream closed due to no active users: "; private static final String STREAM_IN_USE_MSG = "Stream is still in use, remaining users: "; @Resource private ITyService tyService; @@ -48,13 +51,15 @@ webrtcUrl = addVtdu(boxSn); if (Objects.equals(webrtcUrl, "")) { log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn); sendMessageToUser(userId, STREAM_NOT_FOUND_MSG + boxSn); BaseMessage<String> baseMessage = new BaseMessage<>("stream", STREAM_NOT_FOUND_MSG + boxSn); sendMessageToUser(userId, JSON.toJSONString(baseMessage)); } BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl); } else { webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn); } sendMessageToUser(userId, webrtcUrl); BaseMessage<String> baseMessage = new BaseMessage<>("stream", webrtcUrl); sendMessageToUser(userId, JSON.toJSONString(baseMessage)); } /** @@ -68,21 +73,25 @@ Set<Long> users = STREAM_USER_MAP.get(boxSn); if (users == null) { log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, boxSn); sendMessageToUser(userId, "No active users for box: " + boxSn); BaseMessage<String> baseMessage = new BaseMessage<>("stream", "No active users for box: " + boxSn); sendMessageToUser(userId, JSON.toJSONString(baseMessage)); return; } users.remove(userId); String message = ""; if (users.isEmpty()) { mediaService.removePath(boxSn); BOX_STREAM_PULL_MAP.remove(boxSn); STREAM_USER_MAP.remove(boxSn); log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn); sendMessageToUser(userId, STREAM_CLOSED_MSG + boxSn); message = STREAM_CLOSED_MSG + boxSn; } else { log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users); sendMessageToUser(userId, STREAM_IN_USE_MSG + users); message = STREAM_IN_USE_MSG + boxSn; } BaseMessage<String> baseMessage = new BaseMessage<>("stream", message); sendMessageToUser(userId, JSON.toJSONString(baseMessage)); } }
