优化收到redis推动的推流设备信息更新内容重复的问题

648540858

2024-03-13 dc8869be4972b35e4f5a045cb5408ae01f2aad14

 src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java

old mode 100644
new mode 100755
@@ -1,10 +1,10 @@
package com.genersoft.iot.vmp.service.impl;

import com.genersoft.iot.vmp.common.InviteSessionType;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
@@ -47,6 +47,7 @@
 * 设备业务（目录订阅）
 */
@Service
@DS("master")
public class DeviceServiceImpl implements IDeviceService {

    private final static Logger logger = LoggerFactory.getLogger(DeviceServiceImpl.class);
@@ -116,6 +117,7 @@
            inviteStreamService.clearInviteInfo(device.getDeviceId());
        }
        device.setUpdateTime(now);
        device.setKeepaliveTime(now);
        if (device.getKeepaliveIntervalTime() == 0) {
            // 默认心跳间隔60
            device.setKeepaliveIntervalTime(60);
@@ -143,10 +145,6 @@
            }
            sync(device);
        }else {

            if (deviceInDb != null) {
                device.setSwitchPrimarySubStream(deviceInDb.isSwitchPrimarySubStream());
            }
            if(!device.isOnLine()){
                device.setOnLine(true);
                device.setCreateTime(now);
@@ -162,6 +160,19 @@
                    sync(device);
                    // TODO 如果设备下的通道级联到了其他平台，那么需要发送事件或者notify给上级平台
                }
                // 上线添加订阅
                if (device.getSubscribeCycleForCatalog() > 0) {
                    // 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
                    addCatalogSubscribe(device);
                }
                if (device.getSubscribeCycleForMobilePosition() > 0) {
                    addMobilePositionSubscribe(device);
                }
                if (userSetting.getDeviceStatusNotify()) {
                    // 发送redis消息
                    redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
                }

            }else {
                if (deviceChannelMapper.queryAllChannels(device.getDeviceId()).size() == 0) {
                    logger.info("[设备上线]: {}，通道数为0,查询通道信息", device.getDeviceId());
@@ -174,22 +185,21 @@

        }

        // 上线添加订阅
        if (device.getSubscribeCycleForCatalog() > 0) {
            // 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
            addCatalogSubscribe(device);
        }
        if (device.getSubscribeCycleForMobilePosition() > 0) {
            addMobilePositionSubscribe(device);
        }
        // 刷新过期任务
        String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + device.getDeviceId();
        // 如果第一次注册那么必须在60 * 3时间内收到一个心跳，否则设备离线
        dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId(), "首次注册后未能收到心跳"), device.getKeepaliveIntervalTime() * 1000 * 3);
        if (userSetting.getDeviceStatusNotify()) {
            // 发送redis消息
            redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
        }

//
//        try {
//            cmder.alarmSubscribe(device, 600, "0", "4", "0", "2023-7-27T00:00:00", "2023-7-28T00:00:00");
//        } catch (InvalidArgumentException e) {
//            throw new RuntimeException(e);
//        } catch (SipException e) {
//            throw new RuntimeException(e);
//        } catch (ParseException e) {
//            throw new RuntimeException(e);
//        }

    }

@@ -202,6 +212,13 @@
        }
        String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + deviceId;
        dynamicTask.stop(registerExpireTaskKey);
        if (device.isOnLine()) {
            if (userSetting.getDeviceStatusNotify()) {
                // 发送redis消息
                redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, false);
            }
        }

        device.setOnLine(false);
        redisCatchStorage.updateDevice(device);
        deviceMapper.update(device);
@@ -213,16 +230,12 @@
            for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
                mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
                mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
                streamSession.remove(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
                streamSession.removeByCallId(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
            }
        }
        // 移除订阅
        removeCatalogSubscribe(device);
        removeMobilePositionSubscribe(device);
        if (userSetting.getDeviceStatusNotify()) {
            // 发送redis消息
            redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, false);
        }
        removeCatalogSubscribe(device, null);
        removeMobilePositionSubscribe(device, null);

        List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.get(deviceId);
        if (audioBroadcastCatches.size() > 0) {
@@ -260,7 +273,7 @@
    }

    @Override
    public boolean removeCatalogSubscribe(Device device) {
    public boolean removeCatalogSubscribe(Device device, CommonCallback<Boolean> callback) {
        if (device == null || device.getSubscribeCycleForCatalog() < 0) {
            return false;
        }
@@ -270,7 +283,7 @@
            Runnable runnable = dynamicTask.get(taskKey);
            if (runnable instanceof ISubscribeTask) {
                ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
                subscribeTask.stop();
                subscribeTask.stop(callback);
            }
        }
        dynamicTask.stop(taskKey);
@@ -288,12 +301,12 @@
        // 设置最小值为30
        int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
        // 刷新订阅
        dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, (subscribeCycleForCatalog) * 1000);
        dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog * 1000);
        return true;
    }

    @Override
    public boolean removeMobilePositionSubscribe(Device device) {
    public boolean removeMobilePositionSubscribe(Device device, CommonCallback<Boolean> callback) {
        if (device == null || device.getSubscribeCycleForCatalog() < 0) {
            return false;
        }
@@ -303,7 +316,7 @@
            Runnable runnable = dynamicTask.get(taskKey);
            if (runnable instanceof ISubscribeTask) {
                ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
                subscribeTask.stop();
                subscribeTask.stop(callback);
            }
        }
        dynamicTask.stop(taskKey);
@@ -415,8 +428,8 @@
        if (device == null) {
            return null;
        }
        if (ObjectUtils.isEmpty(parentId) || parentId.equals(deviceId)) {
            parentId = null;
        if (ObjectUtils.isEmpty(parentId) ) {
            parentId = deviceId;
        }
        List<DeviceChannel> rootNodes = deviceChannelMapper.getSubChannelsByDeviceId(deviceId, parentId, onlyCatalog);
        return transportChannelsToTree(rootNodes, "");
@@ -494,21 +507,6 @@
            logger.warn("更新设备时未找到设备信息");
            return;
        }
        if(deviceInStore.isSwitchPrimarySubStream() != device.isSwitchPrimarySubStream()){
            //当修改设备的主子码流开关时，需要校验是否存在流，如果存在流则直接关闭
            List<SsrcTransaction> ssrcTransactionForAll = streamSession.getSsrcTransactionForAll(device.getDeviceId(), null, null, null);
            if(ssrcTransactionForAll != null){
                for (SsrcTransaction ssrcTransaction: ssrcTransactionForAll) {
                    try {
                        cmder.streamByeCmd(device, ssrcTransaction.getChannelId(), ssrcTransaction.getStream(), null, null);
                    } catch (InvalidArgumentException | SsrcTransactionNotFoundException | ParseException | SipException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
            deviceChannelMapper.clearPlay(device.getDeviceId());
            inviteStreamService.clearInviteInfo(device.getDeviceId());
        }

        if (!ObjectUtils.isEmpty(device.getName())) {
            deviceInStore.setName(device.getName());
@@ -519,45 +517,87 @@
        if (!ObjectUtils.isEmpty(device.getMediaServerId())) {
            deviceInStore.setMediaServerId(device.getMediaServerId());
        }
        deviceInStore.setSdpIp(device.getSdpIp());
        deviceInStore.setCharset(device.getCharset());

        if (!ObjectUtils.isEmpty(device.getCharset())) {
            deviceInStore.setCharset(device.getCharset());
        }
        if (!ObjectUtils.isEmpty(device.getSdpIp())) {
            deviceInStore.setSdpIp(device.getSdpIp());
        }
        if (!ObjectUtils.isEmpty(device.getPassword())) {
            deviceInStore.setPassword(device.getPassword());
        }
        if (!ObjectUtils.isEmpty(device.getStreamMode())) {
            deviceInStore.setStreamMode(device.getStreamMode());
        }
        //  目录订阅相关的信息
        if (device.getSubscribeCycleForCatalog() > 0) {
            if (deviceInStore.getSubscribeCycleForCatalog() == 0 || deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
                deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
                // 开启订阅
                addCatalogSubscribe(deviceInStore);
            }
        }else if (device.getSubscribeCycleForCatalog() == 0) {
            if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
                deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
                // 取消订阅
                removeCatalogSubscribe(deviceInStore);
            }
        }
        if (deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
            if (device.getSubscribeCycleForCatalog() > 0) {
                // 若已开启订阅，但订阅周期不同，则先取消
                if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
                    removeCatalogSubscribe(deviceInStore, result->{
                        // 开启订阅
                        deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
                        addCatalogSubscribe(deviceInStore);
                        // 因为是异步执行，需要在这里更新下数据
                        deviceMapper.updateCustom(deviceInStore);
                        redisCatchStorage.updateDevice(deviceInStore);
                    });
                }else {
                    // 开启订阅
                    deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
                    addCatalogSubscribe(deviceInStore);
                }

        // 移动位置订阅相关的信息
        if (device.getSubscribeCycleForMobilePosition() > 0) {
            if (deviceInStore.getSubscribeCycleForMobilePosition() == 0 || deviceInStore.getSubscribeCycleForMobilePosition() != device.getSubscribeCycleForMobilePosition()) {
                deviceInStore.setMobilePositionSubmissionInterval(device.getMobilePositionSubmissionInterval());
                deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
                // 开启订阅
                addMobilePositionSubscribe(deviceInStore);
            }
        }else if (device.getSubscribeCycleForMobilePosition() == 0) {
            if (deviceInStore.getSubscribeCycleForMobilePosition() != 0) {
            }else if (device.getSubscribeCycleForCatalog() == 0) {
                // 取消订阅
                removeMobilePositionSubscribe(deviceInStore);
                deviceInStore.setSubscribeCycleForCatalog(0);
                removeCatalogSubscribe(deviceInStore, null);
            }
        }
        // 坐标系变化，需要重新计算GCJ02坐标和WGS84坐标
        if (!deviceInStore.getGeoCoordSys().equals(device.getGeoCoordSys())) {
            updateDeviceChannelGeoCoordSys(device);
        // 移动位置订阅相关的信息
        if (deviceInStore.getSubscribeCycleForMobilePosition() != device.getSubscribeCycleForMobilePosition()) {
            if (device.getSubscribeCycleForMobilePosition() > 0) {
                // 若已开启订阅，但订阅周期不同，则先取消
                if (deviceInStore.getSubscribeCycleForMobilePosition() != 0) {
                    removeMobilePositionSubscribe(deviceInStore, result->{
                        // 开启订阅
                        deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
                        addMobilePositionSubscribe(deviceInStore);
                        // 因为是异步执行，需要在这里更新下数据
                        deviceMapper.updateCustom(deviceInStore);
                        redisCatchStorage.updateDevice(deviceInStore);
                    });
                }else {
                    // 开启订阅
                    deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
                    addMobilePositionSubscribe(deviceInStore);
                }

            }else if (device.getSubscribeCycleForMobilePosition() == 0) {
                // 取消订阅
                deviceInStore.setSubscribeCycleForCatalog(0);
                removeCatalogSubscribe(deviceInStore, null);
            }
        }
        // 更新redis
        redisCatchStorage.updateDevice(device);
        deviceMapper.updateCustom(device);
        if (deviceInStore.getGeoCoordSys() != null) {
            // 坐标系变化，需要重新计算GCJ02坐标和WGS84坐标
            if (!deviceInStore.getGeoCoordSys().equals(device.getGeoCoordSys())) {
                deviceInStore.setGeoCoordSys(device.getGeoCoordSys());
                updateDeviceChannelGeoCoordSys(deviceInStore);
            }
        }else {
            deviceInStore.setGeoCoordSys("WGS84");
        }
        if (device.getCharset() == null) {
            deviceInStore.setCharset("GB2312");
        }
        //SSRC校验
        deviceInStore.setSsrcCheck(device.isSsrcCheck());
        //作为消息通道
        deviceInStore.setAsMessageChannel(device.isAsMessageChannel());

        deviceMapper.updateCustom(deviceInStore);
        redisCatchStorage.updateDevice(deviceInStore);
    }

    @Override