增加websocket订阅无人机视频

liusuyi

2025-10-15 9d77536708f86fb5087c0c6149145066ad98153d

增加websocket订阅无人机视频

 ard-system/src/main/resources/mapper/data/ArdNoFlyZoneMapper.xml

 ard-work/src/main/java/com/ard/work/component/websocket/handler/TyMessageHandler.java

@@ -23,6 +23,8 @@
    @Resource
    private UavControlService uavControlService;
    @Resource
    private UavStreamManager uavStreamManager;
    @Resource
    private UavStateManager uavStateManager;
    @Resource
    private UavTelemetryManager uavTelemetryManager;
@@ -44,8 +46,8 @@
            BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class);
            MessageType messageType = MessageType.fromString(baseMessage.getType());
            switch (messageType) {
                case STREAM: // 直播地址
                    
                case STREAM:
                    handleStreamMessage(payload, userId);
                    break;
                case STATE:
                    handleStateMessage(payload, userId);
@@ -65,6 +67,34 @@
            }
        } catch (Exception e) {
            log.error("Error handling message: {}", message, e);
        }
    }

    /**
     * 处理流消息
     * <p>
     * 该方法解析给定的负载数据，以获取无人机流消息，并根据消息类型决定是否开始或停止流
     *
     * @param payload 消息的负载数据，通常为JSON字符串
     * @param userId  接收消息的用户ID
     */
    private void handleStreamMessage(String payload, Long userId) {
        // 解析JSON负载数据为BaseMessage对象，包含UavStreamMessage类型的有效载荷
        BaseMessage<UavStreamMessage> streamMessage = JSON.parseObject(payload,
                new TypeReference<BaseMessage<UavStreamMessage>>() {
                });
        UavStreamMessage uavStreamMessage = streamMessage.getPayload();

        // 检查解析后的UavStreamMessage对象是否非空
        if (uavStreamMessage != null) {
            String boxSn = uavStreamMessage.getBoxSn();

            // 根据Subscribe决定操作类型：true表示开始拉流，false表示停止拉流
            if (uavStreamMessage.getSubscribe()) {
                uavStreamManager.subscribe(boxSn, userId);
            } else {
                uavStreamManager.unsubscribe(boxSn, userId);
            }
        }
    }

@@ -233,11 +263,9 @@
        WebSocketSessionManager.removeSession(userId);
        log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount());

        if (userId != null) {
            // 释放该用户所有 UAV 控制权
            uavControlService.releaseAllControls(userId);
        }

        uavStreamManager.stopAllPullStreamByUser(userId);
        uavStateManager.removeSubscribe(userId);
        uavTelemetryManager.removeSubscribe(userId);
        uavSignalManager.removeSubscribe(userId);

 ard-work/src/main/java/com/ard/work/component/websocket/handler/UavStreamManager.java

New file
@@ -0,0 +1,131 @@
package com.ard.work.component.websocket.handler;

import com.alibaba.fastjson2.JSON;

import com.ard.vtdu.domain.ArdVtdu;
import com.ard.work.component.websocket.utils.WebSocketUtil;
import com.ard.work.domain.message.BaseMessage;
import com.ard.work.domain.receiver.BoxStreamUrlReceiver;
import com.ard.work.service.ITyService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Objects;
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 = "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;

    // 盒子拉流地址缓存 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<Long>> STREAM_USER_MAP = new ConcurrentHashMap<>();

    /**
     * 开始拉流
     *
     * @param boxSn  盒子序列号
     * @param userId 用户ID
     */
    public void subscribe(String boxSn, Long userId) {

        STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
        String webrtcUrl = getWebrtcUrl(boxSn);
        if (Objects.equals(webrtcUrl, "")) {
            log.warn("{}{}{}", STREAM_LOG_PREFIX, 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);
        BaseMessage<String> baseMessage = new BaseMessage<>("stream", webrtcUrl);
        sendMessageToUser(userId, JSON.toJSONString(baseMessage));
    }

    /**
     * 停止拉流
     *
     * @param boxSn  盒子序列号
     * @param userId 用户ID
     */
    public void unsubscribe(String boxSn, Long userId) {
        synchronized (this) {
            Set<Long> users = STREAM_USER_MAP.get(boxSn);
            if (users == null) {
                log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, 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()) {
                BOX_STREAM_PULL_MAP.remove(boxSn);
                STREAM_USER_MAP.remove(boxSn);
                log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
                message = STREAM_CLOSED_MSG + boxSn;
            } else {
                log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
                message = STREAM_IN_USE_MSG + boxSn;
            }
            BaseMessage<String> baseMessage = new BaseMessage<>("stream", message);
            sendMessageToUser(userId, JSON.toJSONString(baseMessage));
        }
    }

    /**
     * 停止拉流
     *
     * @param userId 用户ID
     */
    public void stopAllPullStreamByUser(Long userId) {
        synchronized (this) {
            STREAM_USER_MAP.forEach((boxSn, users) -> unsubscribe(boxSn, userId));
        }
    }

    /**
     * 发送消息给指定用户
     *
     * @param userId  用户ID
     * @param message 消息内容
     */
    private void sendMessageToUser(Long userId, String message) {
        try {
            WebSocketUtil.sendMessage(userId, message);
        } catch (Exception e) {
            log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e);
        }
    }


    /**
     * 通过盒子编号获取webrtc地址
     *
     * @param boxSn 盒子编号
     * @return String webrtcUrl
     */
    private String getWebrtcUrl(String boxSn) {
        try {
            BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
            if (boxLiveUrl != null) {
                return boxLiveUrl.getRtc();
            }
        } catch (Exception e) {
            log.error("获取媒体地址异常：{}", e.getMessage());
        }
        return "";
    }
}