2025-12-12 zwy 优化，订阅视频时，谁订阅发给谁直播地址，其他订阅不变

zhaowenyu

2025-12-12 7beea5916728a84617d4791775d0aac70ab32eee

2025-12-12 zwy 优化，订阅视频时，谁订阅发给谁直播地址，其他订阅不变

 ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/PropertySetReplyHandler.java

@@ -60,6 +60,6 @@
        String dockControl="";
        dockControl=mapper.writeValueAsString(receiver);

        _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl);
        _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl, "");
    }
}

 ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java

@@ -64,7 +64,7 @@
        String dockControl="";
        dockControl=mapper.writeValueAsString(receiver);

        _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl);
        _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl, "");

//        DockControlProgressWebSockerHandler.
    }

 ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java

@@ -153,6 +153,13 @@
            // 3. 处理订阅/取消订阅逻辑
            handleSubscription(userId, messageType, payload);

            // 直播类型订阅：触发开启直播
            if (MessageType.LIVESTREAM.getType().equals(messageType) && payload.isSubscribe()) {
                String sn = payload.getSn();
                _liveStreamService.StartLiveStreaming(sn,userId);
                log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
            }

        } catch (Exception e) {
            log.error("用户[{}]消息处理异常", userId, e);
            sendMessage(session, "消息格式错误或处理失败：" + e.getMessage());
@@ -187,11 +194,11 @@
            subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
//            log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type);

            // 直播类型订阅：触发开启直播
            if (MessageType.LIVESTREAM.getType().equals(type)) {
                _liveStreamService.StartLiveStreaming(sn);
                log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
            }
//            // 直播类型订阅：触发开启直播
//            if (MessageType.LIVESTREAM.getType().equals(type)) {
//                _liveStreamService.StartLiveStreaming(sn);
//                log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
//            }
        } else {
            // 取消订阅操作
            if (subscriptions.containsKey(subscriptionKey)) {
@@ -257,7 +264,7 @@
     * @param type    消息类型（对应MessageType）
     * @param content 消息内容（JSON字符串）
     */
    public void sendMessageToSubscribers(String sn, String type, String content) {
    public void sendMessageToSubscribers(String sn, String type, String content, String uId) {
        String subscriptionKey = getSubscriptionKey(sn, type);
        Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet());

@@ -280,6 +287,12 @@

            // 线程安全发送消息
            try {
                if(MessageType.LIVESTREAM.getType().equals(type)) {
                    WebSocketSession liveSession = userSessions.get(uId);
                    sendMessage(liveSession, content);
                    log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息：{}", uId, sn, type, content);
                    return;
                }
                sendMessage(session, content);
                log.debug("向用户[{}]推送设备[{}]的[{}]类型消息：{}", userId, sn, type, content);
            } catch (IOException e) {

 ard-sample/src/main/java/com/ard/sample/component/websocket/handler/LiveStreamHandler.java

@@ -31,7 +31,7 @@
    /**
     * 开启直播
     */
    public void StartLiveStreaming(String sn) {
    public void StartLiveStreaming(String sn ,String userId) {
        //调用开始直播
        ResponseResult rlive = null;
        String resultLive = "";
@@ -61,7 +61,7 @@
                    }
                }
            }
            _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), resultLive);
            _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.LIVESTREAM.getType(), resultLive, userId);
            log.info("SN: {}-直播地址：{}", sn, resultLive);
        } catch (Exception e) {
            log.info("SN: {}-开启直播失败！原因：{}", sn, e.getMessage());

 ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java

@@ -191,7 +191,7 @@
            e.printStackTrace();
        }
        // 设备控制进度消息websocket发给前端
        _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.PROGRESS.getType(), dockProgress);
        _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.PROGRESS.getType(), dockProgress, "");

        if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) {
            String topic = headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF;

 ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceOSDServiceImpl.java

@@ -116,7 +116,7 @@
            wsMessage.getData().setHost(data);

            // 发送给websocket客户端
            _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd);
            _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd, "");

            sendMessageService.sendBatch(webSessions, wsMessage);
            this.pushTelemetryData(data, device.getDeviceSn());

 ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java

@@ -1,6 +1,8 @@
package com.ard.sample.manage.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.ard.common.core.domain.model.LoginUser;
import com.ard.common.utils.SecurityUtils;
import com.ard.device.camera.domain.ArdCamera;
import com.ard.device.camera.service.IArdCameraService;
import com.ard.device.channel.domain.ArdChannel;
@@ -369,6 +371,7 @@
    //region 单独写一个方法做权限验证，由于有好多定时任务也在实时获取设备列表，原方法不适用再添加权限注解
    @Override
    public List<DeviceDTO> getDockList(String nickname) {
        LoginUser loginUser = SecurityUtils.getLoginUser();
        List<DeviceDTO> devicesList = this.getDockListByParams(
                DeviceQueryParam.builder()
                        .domains(List.of(DeviceDomainEnum.GATEWAY.getVal(), DeviceDomainEnum.DOCK.getVal()))
@@ -376,8 +379,8 @@
                        .build());
        // 在獲取设备列表时，顺便获取直播地址
        for(DeviceDTO devices : devicesList){
            _liveStreamHandler.StartLiveStreaming(devices.getDeviceSn());
            _liveStreamHandler.StartLiveStreaming(devices.getChildDeviceSn());
            _liveStreamHandler.StartLiveStreaming(devices.getDeviceSn(), loginUser.getUserId().toString());
            _liveStreamHandler.StartLiveStreaming(devices.getChildDeviceSn(), loginUser.getUserId().toString());
            ArdCamera ardCamera = new ArdCamera();
            ardCamera.setDockId(devices.getId());
            List<ArdCamera> ardCameraList = _cameraService.selectArdCameraList(ardCamera);

 ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java

@@ -97,7 +97,7 @@
            wsMessage.getData().setHost(data);
            sendMessageService.sendBatch(webSessions, wsMessage);
            // 发送给websocket客户端
            _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd);
            _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd, "");
        }
    }
}

 ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java

@@ -84,7 +84,7 @@
            String json = getJSONString(MessageType.STATE.getType(), device.getDeviceSn(), dockStatus, droneStatus);

            // 发送给websocket客户端
            _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATE.getType(), json);
            _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATE.getType(), json, "");
        }
    }


 ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java

@@ -149,7 +149,7 @@
        } catch (Exception e) {
            e.printStackTrace();
        }
        _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json);
        _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json, "");
    }

    /**
@@ -194,7 +194,7 @@
        } catch (Exception e) {
            e.printStackTrace();
        }
        _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json);
        _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json, "");
        log.info("定时任务自定义状态为待执行，json: {}", json);
        // endregion
        long now = System.currentTimeMillis();