原始流改为rtsp

liusuyi

2025-02-19 12f344197d6af5d23594847359801c983cf4612a

 ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java

@@ -28,9 +28,9 @@
    @Resource
    private IMediaService mediaService;

    //存储盒子的拉流地址
    private static final ConcurrentMap<String, String> BOX_STREAM_PULL_CACHE = new ConcurrentHashMap<>();
    //存储盒子的拉流用户集合
    //盒子拉流地址缓存
    private static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>();
    //盒子拉流用户集合缓存
    private static final ConcurrentMap<String, Set<String>> STREAM_USER_MAP = new ConcurrentHashMap<>();

    @Override
@@ -48,29 +48,28 @@
            }
        }

        log.info("【stream】Received message: " + payload);
        log.info("【stream】size: " + WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
        log.info("【stream】Received message: {}", payload);
        log.info("【stream】size: {}", WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
    }

    //开始拉流
    private void startPullStream(String boxSn, String userId) {
        //BOX_STREAM_USER_COUNT.merge(boxSn, 1, Integer::sum);
        STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
        String messageReply;
        if (BOX_STREAM_PULL_CACHE.containsKey(boxSn)) {
            String webrtcUrl = BOX_STREAM_PULL_CACHE.get(boxSn);
        if (BOX_STREAM_PULL_MAP.containsKey(boxSn)) {
            String webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn);
            messageReply = webrtcUrl;
            log.warn("【stream】Already pulled stream: " + webrtcUrl);
            log.warn("【stream】Already pulled stream: {}", webrtcUrl);
        } else {
            BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
            if (boxLiveUrl != null) {
                ArdVtdu vtdu = ArdVtdu.builder().name(boxSn).mode(0).isCode(0).rtspSource(boxLiveUrl.getRtmp()).build();
                ArdVtdu vtdu = ArdVtdu.builder().name(boxSn).mode(0).isCode(0).rtspSource(boxLiveUrl.getRtsp()).build();
                ArdVtdu ardVtdu = mediaService.addPath(vtdu);
                BOX_STREAM_PULL_CACHE.put(boxSn, ardVtdu.getWebrtcUrl());
                BOX_STREAM_PULL_MAP.put(boxSn, ardVtdu.getWebrtcUrl());
                messageReply = ardVtdu.getWebrtcUrl();
                log.warn("【stream】First pulled stream: " + ardVtdu.getWebrtcUrl());
                log.warn("【stream】First pulled stream: {}", ardVtdu.getWebrtcUrl());
            } else {
                log.warn("【stream】Box stream not found: " + boxSn);
                log.warn("【stream】Box stream not found: {}", boxSn);
                messageReply = "【stream】Box stream not found: " + boxSn;
            }
        }
@@ -99,17 +98,17 @@
    //关闭拉流
    private void closeStream(String boxSn) {
        mediaService.removePath(boxSn);
        BOX_STREAM_PULL_CACHE.remove(boxSn);
        BOX_STREAM_PULL_MAP.remove(boxSn);
        STREAM_USER_MAP.remove(boxSn);
        log.warn("【stream】Stream closed due to no active users: " + boxSn);
        log.warn("【stream】Stream closed due to no active users: {}", boxSn);
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String userId = (String) session.getAttributes().get("userId");
        WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.put(userId, session);
        log.info("【stream】Connected: " + session.getId());
        log.info("【stream】size: " + WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
        log.info("【stream】Connected: {}", session.getId());
        log.info("【stream】size: {}", WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
    }

    @Override
@@ -117,7 +116,7 @@
        String userId = (String) session.getAttributes().get("userId");
        WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.remove(userId);
        STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
        log.info("【stream】Disconnected: " + session.getId());
        log.info("【stream】size: " + WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
        log.info("【stream】Disconnected: {}", session.getId());
        log.info("【stream】size: {}", WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
    }
}