forked from uav/ard-uav-server
ard-work/src/main/java/com/ard/tianyu/aspect/ControlAuthorityAspect.java
@@ -57,7 +57,7 @@ String boxSn = getBoxSn(arg); if (boxSn != null) { //判断无人机控制者是否为当前登录用户 Long controller = UavControlManager.droneControlMap.get(boxSn); Long controller = UavControlManager.uavControlMap.get(boxSn); if (controller != null && controller.equals(SecurityUtils.getUserId())) { return true; } ard-work/src/main/java/com/ard/tianyu/controller/UavPermissionsController.java
@@ -52,6 +52,6 @@ @ApiOperationSupport(order = 4) @GetMapping("/controlRequestMap") public AjaxResult controlRequestMap(String boxSn) { return AjaxResult.success(UavControlManager.droneControlMap.get(boxSn)); return AjaxResult.success(UavControlManager.uavControlMap.get(boxSn)); } } ard-work/src/main/java/com/ard/tianyu/service/impl/UavControlManager.java
@@ -3,94 +3,156 @@ import com.alibaba.fastjson2.JSON; import com.ard.common.exception.NoPermissionException; import com.ard.common.utils.SecurityUtils; import com.ard.tianyu.websocket.domain.ControlMessage; import com.ard.tianyu.websocket.domain.WsMessage; import com.ard.tianyu.websocket.domain.ControlAuthorityDTO; import com.ard.tianyu.websocket.domain.BaseMessage; import com.ard.tianyu.websocket.utils.WebSocketUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import java.util.concurrent.ConcurrentHashMap; /** * 无人机控制管理器 * * @author liusuyi * @date 2024年02月21日8:37 **/ @Service @Slf4j public class UavControlManager { // 存储每个无人机的当前控制者 public static final ConcurrentHashMap<String, Long> droneControlMap = new ConcurrentHashMap<>(); private static final String CONTROL_REQUEST_ACTION = "controlRequest"; private static final String REQUEST_ACTION = "request"; private static final String APPROVE_ACTION = "approve"; private static final String REJECT_ACTION = "reject"; // 存储每个无人机的当前控制者 public static final ConcurrentHashMap<String, Long> uavControlMap = new ConcurrentHashMap<>(); // 存储无人机的控制请求（用于判断哪个控制者正在等待） public static final ConcurrentHashMap<String, Long> controlRequestMap = new ConcurrentHashMap<>(); // 发送通知（通过 WebSocket推送） public void notifyMessage(String boxSn, String action, Long source, Long target) { ControlMessage controlMessage = ControlMessage.builder() .boxSn(boxSn) .userId(source) .action(action) .build(); WsMessage<ControlMessage> wsMessage = new WsMessage<>("controlRequest", controlMessage); WebSocketUtil.sendMessage(target.toString(), JSON.toJSONString(wsMessage)); } // 申请控制 /** * 申请控制 * * @param boxSn 盒子编号 */ public boolean requestControl(String boxSn) { Long requester = SecurityUtils.getUserId(); Long controller = uavControlMap.get(boxSn); // 如果无人机没有控制者，直接控制 if (droneControlMap.get(boxSn) == null) { droneControlMap.put(boxSn, requester); if (controller == null) { uavControlMap.put(boxSn, requester); log.info("控制权已授予给用户: {}", requester); return true; // 控制成功 } // 如果无人机已经有控制者,则发送申请给当前控制者 Long controller = droneControlMap.get(boxSn); if (!controller.equals(requester)) { // 则将控制请求加入请求队列 controlRequestMap.put(boxSn, requester); // 通知当前控制者，有新的控制请求 notifyMessage(boxSn, "request", requester, controller); notifyMessage(boxSn, REQUEST_ACTION, requester, controller); log.info("控制请求已发送给当前控制者: {}", controller); } else { throw new NoPermissionException("当前已经获取控制权，无需重复申请"); } return false; // 等待当前控制者的同意 } // 处理控制请求的同意或拒绝 /** * 处理控制请求 * * @param boxSn 盒子编号 * @param approve 是否同意 */ public boolean handleControlResponse(String boxSn, boolean approve) { Long controller = SecurityUtils.getUserId(); Long requester = controlRequestMap.get(boxSn); // 确保请求者是当前的控制者 if (requester != null && !requester.equals(controller)) { if (approve) { // 同意控制，将控制权赋予请求者 droneControlMap.put(boxSn, requester); notifyMessage(boxSn, "approve", controller, requester); } else { notifyMessage(boxSn, "reject", controller, requester); if (requester == null || requester.equals(controller)) { throw new NoPermissionException("无申请不需要处理控制权"); } // 无论同意还是拒绝，都移除控制请求 if (approve) { uavControlMap.put(boxSn, requester); notifyMessage(boxSn, APPROVE_ACTION, controller, requester); log.info("控制权已转移给用户: {}", requester); } else { notifyMessage(boxSn, REJECT_ACTION, controller, requester); log.info("控制请求被拒绝，请求者: {}", requester); } controlRequestMap.remove(boxSn); return true; } else { throw new NoPermissionException("无申请不需要处理控制权"); } } // 释放控制 /** * 释放控制 * * @param boxSn 盒子编号 */ public boolean releaseControl(String boxSn) { Long controller = droneControlMap.get(boxSn); if (controller != null) { if (controller.equals(SecurityUtils.getUserId())) { droneControlMap.remove(boxSn); return true; } else { // 当前控制者不匹配，可以记录日志或者抛出异常 log.warn("无权限释放控制, 实际控制者是 {}", controller); throw new NoPermissionException("无权限释放控制, 实际控制者是 " + controller); } } else { // 说明无人机没有控制者 Long controller = uavControlMap.get(boxSn); if (controller == null) { log.warn("无人机当前 {} 没有控制者,无需释放", boxSn); throw new NoPermissionException("无人机" + boxSn + "当前没有控制者,无需释放"); } if (!controller.equals(SecurityUtils.getUserId())) { log.warn("无权限释放控制, 实际控制者是 {}", controller); throw new NoPermissionException("无权限释放控制, 实际控制者是 " + controller); } uavControlMap.remove(boxSn); log.info("控制权已释放，无人机: {}", boxSn); return true; } /** * 释放指定用户控制的无人机 * * @param userId 用户id */ public void releaseAllControlByUser(String userId) { String boxSn = uavControlMap.entrySet().stream() .filter(entry -> entry.getValue().toString().equals(userId)) .map(ConcurrentHashMap.Entry::getKey) .findFirst() .orElse(null); if (boxSn != null) { // 用户控制了某个无人机，移除缓存 uavControlMap.remove(boxSn); log.info("移除{}控制的无人机: {}", userId, boxSn); } } /** * 发送消息给指定用户 * * @param boxSn 盒子编号 * @param action 指令() * @param source 发送方用户ID * @param target 接收方用户ID */ private void notifyMessage(String boxSn, String action, Long source, Long target) { ControlAuthorityDTO controlAuthorityDTO = buildControlAuthorityDTO(boxSn, action, source); BaseMessage<ControlAuthorityDTO> baseMessage = buildBaseMessage(CONTROL_REQUEST_ACTION, controlAuthorityDTO); WebSocketUtil.sendMessage(target.toString(), JSON.toJSONString(baseMessage)); } private ControlAuthorityDTO buildControlAuthorityDTO(String boxSn, String action, Long userId) { return ControlAuthorityDTO.builder() .boxSn(boxSn) .userId(userId) .action(action) .build(); } private <T> BaseMessage<T> buildBaseMessage(String type, T data) { return new BaseMessage<>(type, data); } } ard-work/src/main/java/com/ard/tianyu/service/impl/UavStreamManager.java
New file @@ -0,0 +1,129 @@ package com.ard.tianyu.service.impl; import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver; import com.ard.tianyu.service.ITyService; import com.ard.tianyu.websocket.utils.WebSocketUtil; import com.ard.vtdu.domain.ArdVtdu; import com.ard.vtdu.service.IMediaService; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @Service @Slf4j 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: "; @Resource private ITyService tyService; @Resource private IMediaService mediaService; // 盒子拉流地址缓存 key:boxSn value: webrtcUrl public static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>(); // 盒子拉流用户集合缓存 key:boxSn value:Set<UserId> public static final ConcurrentMap<String, Set<String>> STREAM_USER_MAP = new ConcurrentHashMap<>(); /** * 开始拉流 * * @param boxSn 盒子序列号 * @param userId 用户ID */ public void startPullStream(String boxSn, String userId) { STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId); String webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn); if (webrtcUrl != null) { log.info("{}Already pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl); sendMessageToUser(userId, webrtcUrl); return; } try { BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn); if (boxLiveUrl == null) { log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn); sendMessageToUser(userId, STREAM_NOT_FOUND_MSG + boxSn); return; } ArdVtdu vtdu = ArdVtdu.builder() .name(boxSn) .mode(0) .isCode(0) .rtspSource(boxLiveUrl.getRtsp()) .build(); ArdVtdu ardVtdu = mediaService.addPath(vtdu); webrtcUrl = ardVtdu.getWebrtcUrl(); BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl); log.info("{}First pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl); sendMessageToUser(userId, webrtcUrl); } catch (Exception e) { log.error("{}Failed to start pull stream for box: {}", STREAM_LOG_PREFIX, boxSn, e); sendMessageToUser(userId, "Failed to start pull stream due to an internal error."); } } /** * 停止拉流 * * @param boxSn 盒子序列号 * @param userId 用户ID */ public void stopPullStream(String boxSn, String userId) { synchronized (this) { Set<String> 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); return; } users.remove(userId); 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); } else { log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users); sendMessageToUser(userId, STREAM_IN_USE_MSG + users); } } } /** * 停止拉流 * * @param userId 用户ID */ public void stopAllPullStreamByUser(String userId) { synchronized (this) { STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId)); } } /** * 发送消息给指定用户 * * @param userId 用户ID * @param message 消息内容 */ private void sendMessageToUser(String userId, String message) { try { WebSocketUtil.sendMessage(userId, message); } catch (Exception e) { log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e); } } } ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java
@@ -1,12 +1,12 @@ package com.ard.tianyu.websocket.config; import com.ard.tianyu.websocket.handler.TyMessageHandler; import com.ard.tianyu.websocket.handler.TyStreamHandler; import com.ard.tianyu.websocket.interceptor.TyHandshakeInterceptor; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; import javax.annotation.Resource; @Configuration @@ -15,20 +15,15 @@ @Resource private TyMessageHandler tyMessageHandler; @Resource private TyStreamHandler tyStreamHandler; @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { //添加myWebSocketHandler消息处理对象以及websocket连接地址 registry.addHandler(tyMessageHandler, "/ws/control/**") registry.addHandler(tyMessageHandler, "/ws/**") //设置允许跨域访问 .setAllowedOrigins("*") //添加拦截器可实现用户链接前进行权限校验等操作 .addInterceptors(new TyHandshakeInterceptor()); // 注册流媒体 WebSocket 处理器 registry.addHandler(tyStreamHandler, "/ws/stream/**") .setAllowedOrigins("*") .addInterceptors(new TyHandshakeInterceptor()); // 可选的拦截器 } } ard-work/src/main/java/com/ard/tianyu/websocket/domain/BaseMessage.java
File was renamed from ard-work/src/main/java/com/ard/tianyu/websocket/domain/WsMessage.java @@ -7,7 +7,7 @@ @Data @AllArgsConstructor @NoArgsConstructor public class WsMessage<T> { public class BaseMessage<T> { private String type; // 消息类型 private T payload; // 泛型消息内容 } ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlAuthorityDTO.java
File was renamed from ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlMessage.java @@ -9,7 +9,7 @@ @Builder @AllArgsConstructor @NoArgsConstructor public class ControlMessage { public class ControlAuthorityDTO { private String boxSn; // 盒子编号 private Long userId; // 消息发送者 ard-work/src/main/java/com/ard/tianyu/websocket/domain/PullStreamDTO.java
File was renamed from ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java @@ -1,4 +1,4 @@ package com.ard.tianyu.websocket.model; package com.ard.tianyu.websocket.domain; import lombok.AllArgsConstructor; import lombok.Builder; ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
@@ -1,5 +1,11 @@ package com.ard.tianyu.websocket.handler; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.TypeReference; import com.ard.tianyu.service.impl.UavControlManager; import com.ard.tianyu.service.impl.UavStreamManager; import com.ard.tianyu.websocket.domain.BaseMessage; import com.ard.tianyu.websocket.domain.PullStreamDTO; import com.ard.tianyu.websocket.utils.WebSocketUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -7,17 +13,42 @@ import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import javax.annotation.Resource; @Component @Slf4j public class TyMessageHandler extends TextWebSocketHandler { @Resource private UavControlManager uavControlManager; @Resource private UavStreamManager uavStreamManager; @Override public void handleTextMessage(WebSocketSession session, TextMessage message) { // 处理接收到的消息 String userId = (String) session.getAttributes().get("userId"); String payload = message.getPayload(); log.info("【cmd】Received message: {}", payload); log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size()); BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class); switch (baseMessage.getType()) { case "stream": BaseMessage<PullStreamDTO> streamMessage = JSON.parseObject(payload, new TypeReference<BaseMessage<PullStreamDTO>>() { }); PullStreamDTO pullStreamDTO = streamMessage.getPayload(); if (pullStreamDTO != null) { String boxSn = pullStreamDTO.getBoxSn(); if (pullStreamDTO.getPullType().equals(1)) { uavStreamManager.startPullStream(boxSn, userId); } else { uavStreamManager.stopPullStream(boxSn, userId); } } break; } } @Override @@ -35,5 +66,9 @@ String userId = (String) session.getAttributes().get("userId"); WebSocketUtil.ONLINE_USER_SESSIONS.remove(userId); log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size()); //停止用户的所有流 uavStreamManager.stopAllPullStreamByUser(userId); //移除控制的无人机 uavControlManager.releaseAllControlByUser(userId); } } ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java
File was deleted ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java
@@ -27,9 +27,6 @@ // 存储 websocket session public static final ConcurrentMap<String, WebSocketSession> ONLINE_USER_SESSIONS = new ConcurrentHashMap<>(); // 存储 拉流 websocket session public static final ConcurrentMap<String, WebSocketSession> STREAM_ONLINE_USER_SESSIONS = new ConcurrentHashMap<>(); /** * @param session 用户 session * @param message 发送内容 @@ -51,10 +48,12 @@ } } } public static void sendMessage(String userId, String message) { WebSocketSession session = ONLINE_USER_SESSIONS.get(userId); sendMessage(session, message); } /** * @param session 用户 session * @param message 发送内容 @@ -101,6 +100,7 @@ public static void sendMessageAll(String message) { ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, message)); } /** * 推送消息到其他客户端 * @@ -110,6 +110,7 @@ JSONObject jsonObject = new JSONObject(message); ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, jsonObject.toString())); } /** * 推送消息到其他客户端 * @@ -119,6 +120,7 @@ String jsonMessage = JSONObject.toJSONString(message); // 确保对象转换为 JSON 字符串 ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, jsonMessage)); } /** * 推送消息到其他客户端 * @@ -127,6 +129,7 @@ public static void sendMessageAll(List message) { ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, JSON.toJSONString(message))); } /** * 发送消息给userId_前缀的人 */ ard-work/src/main/java/com/ard/tianyu/websocket/utils/WsMessageConverter.java
File was deleted
