优化订阅机制，需要重新订阅时，取消命令发送后再发送订阅命令 #1273

648540858

2024-02-20 a9ab5c28e9fd52c1d936a245ac46c9e556f6bc3e

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

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.service.impl;

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;
@@ -231,8 +232,8 @@
            }
        }
        // 移除订阅
        removeCatalogSubscribe(device);
        removeMobilePositionSubscribe(device);
        removeCatalogSubscribe(device, null);
        removeMobilePositionSubscribe(device, null);
    }

    @Override
@@ -251,7 +252,7 @@
    }

    @Override
    public boolean removeCatalogSubscribe(Device device) {
    public boolean removeCatalogSubscribe(Device device, CommonCallback<Boolean> callback) {
        if (device == null || device.getSubscribeCycleForCatalog() < 0) {
            return false;
        }
@@ -261,7 +262,7 @@
            Runnable runnable = dynamicTask.get(taskKey);
            if (runnable instanceof ISubscribeTask) {
                ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
                subscribeTask.stop();
                subscribeTask.stop(callback);
            }
        }
        dynamicTask.stop(taskKey);
@@ -284,7 +285,7 @@
    }

    @Override
    public boolean removeMobilePositionSubscribe(Device device) {
    public boolean removeMobilePositionSubscribe(Device device, CommonCallback<Boolean> callback) {
        if (device == null || device.getSubscribeCycleForCatalog() < 0) {
            return false;
        }
@@ -294,7 +295,7 @@
            Runnable runnable = dynamicTask.get(taskKey);
            if (runnable instanceof ISubscribeTask) {
                ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
                subscribeTask.stop();
                subscribeTask.stop(callback);
            }
        }
        dynamicTask.stop(taskKey);
@@ -522,39 +523,54 @@
        if (!ObjectUtils.isEmpty(device.getStreamMode())) {
            deviceInStore.setStreamMode(device.getStreamMode());
        }


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

            }else if (device.getSubscribeCycleForCatalog() == 0) {
                // 取消订阅
                deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
                removeCatalogSubscribe(deviceInStore);
                deviceInStore.setSubscribeCycleForCatalog(0);
                removeCatalogSubscribe(deviceInStore, null);
            }
        }

        // 移动位置订阅相关的信息
        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) {
                deviceInStore.setMobilePositionSubmissionInterval(device.getMobilePositionSubmissionInterval());
                deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
        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) {
                // 取消订阅
                removeMobilePositionSubscribe(deviceInStore);
                deviceInStore.setSubscribeCycleForCatalog(0);
                removeCatalogSubscribe(deviceInStore, null);
            }
        }
        if (deviceInStore.getGeoCoordSys() != null) {
@@ -574,9 +590,8 @@
        //作为消息通道
        deviceInStore.setAsMessageChannel(device.isAsMessageChannel());
        
        // 更新redis
        deviceMapper.updateCustom(deviceInStore);
        redisCatchStorage.removeDevice(deviceInStore.getDeviceId());
        redisCatchStorage.updateDevice(deviceInStore);
    }

    @Override