异步通道刷新，优化ui效果

648540858

2022-04-12 3955e6ed53d450f8faf488d4b74ba0c0c83c5aaa

异步通道刷新，优化ui效果

 src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java

@@ -22,6 +22,9 @@


    public static final String DEVICE_PREFIX = "VMP_DEVICE_";



    // 设备同步完成

    public static final String DEVICE_SYNC_PREFIX = "VMP_DEVICE_SYNC_";



    public static final String CACHEKEY_PREFIX = "VMP_CHANNEL_";



    public static final String KEEPLIVEKEY_PREFIX = "VMP_KEEPALIVE_";

@@ -69,6 +72,7 @@


    public static final String SYSTEM_INFO_NET_PREFIX = "VMP_SYSTEM_INFO_NET_";





    //************************** redis 消息*********************************



    // 流变化的通知


 src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java

@@ -8,6 +8,12 @@
    private List<DeviceChannel> channelList;
    private Date lastTime;
    private Device device;
    private String errorMsg;

    public enum CatalogDataStatus{
        ready, runIng, end
    }
    private CatalogDataStatus status;

    public int getTotal() {
        return total;
@@ -40,4 +46,20 @@
    public void setDevice(Device device) {
        this.device = device;
    }

    public String getErrorMsg() {
        return errorMsg;
    }

    public void setErrorMsg(String errorMsg) {
        this.errorMsg = errorMsg;
    }

    public CatalogDataStatus getStatus() {
        return status;
    }

    public void setStatus(CatalogDataStatus status) {
        this.status = status;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/SyncStatus.java

New file
@@ -0,0 +1,34 @@
package com.genersoft.iot.vmp.gb28181.bean;

/**
 * 摄像机同步状态
 */
public class SyncStatus {
    private int total;
    private int current;
    private String errorMsg;

    public int getTotal() {
        return total;
    }

    public void setTotal(int total) {
        this.total = total;
    }

    public int getCurrent() {
        return current;
    }

    public void setCurrent(int current) {
        this.current = current;
    }

    public String getErrorMsg() {
        return errorMsg;
    }

    public void setErrorMsg(String errorMsg) {
        this.errorMsg = errorMsg;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/event/online/OnlineEventListener.java

@@ -97,8 +97,6 @@
        }

        // 处理上线监听

        storager.updateDevice(device);

        List<DeviceChannel> deviceChannelList = storager.queryOnlineChannelsByDeviceId(device.getDeviceId());

        eventPublisher.catalogEventPublish(null, deviceChannelList, CatalogEvent.ON);

        // 上线添加订阅

        if (device.getSubscribeCycleForCatalog() > 0) {

            deviceService.addCatalogSubscribe(device);


 src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java

@@ -3,6 +3,7 @@
import com.genersoft.iot.vmp.gb28181.bean.CatalogData;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -25,6 +26,17 @@
    @Autowired
    private IVideoManagerStorage storager;

    public void addReady(String key) {
        CatalogData catalogData = data.get(key);
        if (catalogData == null || catalogData.getStatus().equals(CatalogData.CatalogDataStatus.end)) {
            catalogData = new CatalogData();
            catalogData.setChannelList(new ArrayList<>());
            catalogData.setStatus(CatalogData.CatalogDataStatus.ready);
            catalogData.setLastTime(new Date(System.currentTimeMillis()));
            data.put(key, catalogData);
        }
    }

    public void put(String key, int total, Device device, List<DeviceChannel> deviceChannelList) {
        CatalogData catalogData = data.get(key);
        if (catalogData == null) {
@@ -32,10 +44,16 @@
            catalogData.setTotal(total);
            catalogData.setDevice(device);
            catalogData.setChannelList(new ArrayList<>());
            catalogData.setStatus(CatalogData.CatalogDataStatus.runIng);
            catalogData.setLastTime(new Date(System.currentTimeMillis()));
            data.put(key, catalogData);
        }
        }else {
            catalogData.setTotal(total);
            catalogData.setDevice(device);
            catalogData.setStatus(CatalogData.CatalogDataStatus.runIng);
        catalogData.getChannelList().addAll(deviceChannelList);
        catalogData.setLastTime(new Date(System.currentTimeMillis()));
        }
    }

    public List<DeviceChannel> get(String key) {
@@ -50,6 +68,16 @@
        return catalogData.getTotal();
    }

    public SyncStatus getSyncStatus(String key) {
        CatalogData catalogData = data.get(key);
        if (catalogData == null) return null;
        SyncStatus syncStatus = new SyncStatus();
        syncStatus.setCurrent(catalogData.getChannelList().size());
        syncStatus.setTotal(catalogData.getTotal());
        syncStatus.setErrorMsg(catalogData.getErrorMsg());
        return syncStatus;
    }

    public void del(String key) {
        data.remove(key);
    }
@@ -57,24 +85,32 @@
    @Scheduled(fixedRate = 5 * 1000)   //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时
    private void timerTask(){
        Set<String> keys = data.keySet();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(new Date());
        calendar.set(Calendar.SECOND, calendar.get(Calendar.SECOND) - 5);
        Calendar calendarBefore5S = Calendar.getInstance();
        calendarBefore5S.setTime(new Date());
        calendarBefore5S.set(Calendar.SECOND, calendarBefore5S.get(Calendar.SECOND) - 5);

        Calendar calendarBefore30S = Calendar.getInstance();
        calendarBefore30S.setTime(new Date());
        calendarBefore30S.set(Calendar.SECOND, calendarBefore30S.get(Calendar.SECOND) - 30);
        for (String key : keys) {
            CatalogData catalogData = data.get(key);
            if (catalogData.getLastTime().before(calendar.getTime())) {

            if (catalogData.getLastTime().before(calendarBefore5S.getTime())) { // 超过五秒收不到消息任务超时， 只更新这一部分数据
                storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
                RequestMessage msg = new RequestMessage();
                msg.setKey(key);
                WVPResult<Object> result = new WVPResult<>();
                result.setCode(0);
                result.setMsg("更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条");
                result.setData(catalogData.getDevice());
                msg.setData(result);
                deferredResultHolder.invokeAllResult(msg);
                String errorMsg = "更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条";
                catalogData.setStatus(CatalogData.CatalogDataStatus.end);
                catalogData.setErrorMsg(errorMsg);
            }
            if (catalogData.getLastTime().before(calendarBefore30S.getTime())) { // 超过三十秒，如果标记为end则删除
                data.remove(key);
            }
        }
    }


    public void setChannelSyncEnd(String key, String errorMsg) {
        CatalogData catalogData = data.get(key);
        if (catalogData == null)return;
        catalogData.setStatus(CatalogData.CatalogDataStatus.end);
        catalogData.setErrorMsg(errorMsg);
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java

@@ -46,6 +46,7 @@
     * @return
     */
    boolean catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size);
    boolean catalogQuery(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag);

    /**
     * 向上级回复DeviceInfo查询信息

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -215,6 +215,31 @@
            return false;
        }
        try {
            String catalogXml = getCatalogXml(channel, sn, parentPlatform, size);

            // callid
            CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                    : udpSipProvider.getNewCallId();

            Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, catalogXml.toString(), fromTag, callIdHeader);
            transmitRequest(parentPlatform, request);

        } catch (SipException | ParseException | InvalidArgumentException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Override
    public boolean catalogQuery(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag) {
        if ( parentPlatform ==null) {
            return false;
        }
        sendCatalogResponse(channels, parentPlatform, sn, fromTag, 0);
        return true;
    }
    private String getCatalogXml(DeviceChannel channel, String sn, ParentPlatform parentPlatform, int size) {
            String characterSet = parentPlatform.getCharacterSet();
            StringBuffer catalogXml = new StringBuffer(600);
            catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet +"\" ?>\r\n");
@@ -253,19 +278,28 @@
            catalogXml.append("</Item>\r\n");
            catalogXml.append("</DeviceList>\r\n");
            catalogXml.append("</Response>\r\n");
        return catalogXml.toString();
    }

    private void sendCatalogResponse(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag, int index) {
        if (index >= channels.size()) {
            return;
        }
        try {
            DeviceChannel deviceChannel = channels.get(index);
            String catalogXml = getCatalogXml(deviceChannel, sn, parentPlatform, channels.size());
            // callid
            CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                    : udpSipProvider.getNewCallId();

            Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, catalogXml.toString(), fromTag, callIdHeader);
            transmitRequest(parentPlatform, request);

            Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, catalogXml, fromTag, callIdHeader);
            transmitRequest(parentPlatform, request, null, eventResult -> {
                int indexNext = index + 1;
                sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext);
            });
        } catch (SipException | ParseException | InvalidArgumentException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/MessageRequestProcessor.java

@@ -12,6 +12,7 @@
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.slf4j.Logger;
@@ -23,6 +24,7 @@
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CSeqHeader;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
@@ -81,6 +83,17 @@
        // 查询上级平台是否存在
        ParentPlatform parentPlatform = storage.queryParentPlatByServerGBId(deviceId);
        try {
            if (device != null && parentPlatform != null) {
                logger.warn("[重复]平台与设备编号重复：{}", deviceId);
                SIPRequest request = (SIPRequest) evt.getRequest();
                String hostAddress = request.getRemoteAddress().getHostAddress();
                int remotePort = request.getRemotePort();
                if (device.getHostAddress().equals(hostAddress + ":" + remotePort)) {
                    parentPlatform = null;
                }else {
                    device = null;
                }
            }
            if (device == null && parentPlatform == null) {
                // 不存在则回复404
                responseAck(evt, Response.NOT_FOUND, "device "+ deviceId +" not found");

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/CatalogNotifyMessageHandler.java

@@ -18,6 +18,7 @@
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;

@Component
@@ -58,7 +59,8 @@
            List<DeviceChannelInPlatform> deviceChannels = storage.queryChannelListInParentPlatform(parentPlatform.getServerGBId());
            // 查询关联的直播通道
            List<GbStream> gbStreams = storage.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
            int size = deviceChannels.size() + gbStreams.size();

            List<DeviceChannel> allChannels = new ArrayList<>();
            // 回复目录信息
            List<PlatformCatalog> catalogs =  storage.queryCatalogInPlatform(parentPlatform.getServerGBId());
            if (catalogs.size() > 0) {
@@ -81,9 +83,7 @@
                    deviceChannel.setModel("live");
                    deviceChannel.setOwner("wvp-pro");
                    deviceChannel.setSecrecy("0");
                    cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                    // 防止发送过快
                    Thread.sleep(100);
                    allChannels.add(deviceChannel);
                }
            }
            // 回复级联的通道
@@ -96,9 +96,7 @@
                    deviceChannel.setParental(0);
                    deviceChannel.setParentId(channel.getCatalogId());
                    deviceChannel.setCivilCode(parentPlatform.getDeviceGBId().substring(0, 6));
                    cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                    // 防止发送过快
                    Thread.sleep(100);
                    allChannels.add(deviceChannel);
                }
            }
            // 回复直播的通道
@@ -123,16 +121,16 @@
                    deviceChannel.setOwner("wvp-pro");
                    deviceChannel.setParental(0);
                    deviceChannel.setSecrecy("0");
                    cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                    // 防止发送过快
                    Thread.sleep(100);
                    allChannels.add(deviceChannel);
                }
            }
            if (size == 0) {
            if (allChannels.size() > 0) {
                cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag());
            }else {
                // 回复无通道
                cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), size);
                cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0);
            }
        } catch (SipException | InvalidArgumentException | ParseException | InterruptedException e) {
        } catch (SipException | InvalidArgumentException | ParseException e) {
            e.printStackTrace();
        }


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java

@@ -22,6 +22,7 @@
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;

@Component
@@ -44,6 +45,9 @@

    @Autowired
    private EventPublisher publisher;

    @Autowired
    private IVideoManagerStorage storage;

    @Override
    public void afterPropertiesSet() throws Exception {
@@ -71,10 +75,11 @@
            List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
            // 回复目录信息
            List<PlatformCatalog> catalogs =  storager.queryCatalogInPlatform(parentPlatform.getServerGBId());
            int size = catalogs.size() + deviceChannelInPlatforms.size() + gbStreams.size();

            List<DeviceChannel> allChannels = new ArrayList<>();
            if (catalogs.size() > 0) {
                for (PlatformCatalog catalog : catalogs) {
                    if (catalog.getParentId().equals(parentPlatform.getServerGBId())) {
                    if (catalog.getParentId().equals(catalog.getPlatformId())) {
                        catalog.setParentId(parentPlatform.getDeviceGBId());
                    }
                    DeviceChannel deviceChannel = new DeviceChannel();
@@ -92,9 +97,7 @@
                    deviceChannel.setModel("live");
                    deviceChannel.setOwner("wvp-pro");
                    deviceChannel.setSecrecy("0");
                    cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                    // 防止发送过快
                    Thread.sleep(100);
                    allChannels.add(deviceChannel);
                }
            }
            // 回复级联的通道
@@ -103,20 +106,18 @@
                    if (channel.getCatalogId().equals(parentPlatform.getServerGBId())) {
                        channel.setCatalogId(parentPlatform.getDeviceGBId());
                    }
                    DeviceChannel deviceChannel = storager.queryChannel(channel.getDeviceId(), channel.getChannelId());
                    DeviceChannel deviceChannel = storage.queryChannel(channel.getDeviceId(), channel.getChannelId());
                    deviceChannel.setParental(0);
                    deviceChannel.setParentId(channel.getCatalogId());
                    deviceChannel.setCivilCode(parentPlatform.getDeviceGBId().substring(0, 6));
                    cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                    // 防止发送过快
                    Thread.sleep(100);
                    allChannels.add(deviceChannel);
                }
            }
            // 回复直播的通道
            if (gbStreams.size() > 0) {
                for (GbStream gbStream : gbStreams) {
                    if (gbStream.getCatalogId().equals(parentPlatform.getServerGBId())) {
                        gbStream.setCatalogId(parentPlatform.getDeviceGBId());
                        gbStream.setCatalogId(null);
                    }
                    DeviceChannel deviceChannel = new DeviceChannel();
                    deviceChannel.setChannelId(gbStream.getGbId());
@@ -134,23 +135,20 @@
                    deviceChannel.setOwner("wvp-pro");
                    deviceChannel.setParental(0);
                    deviceChannel.setSecrecy("0");

                    cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                    // 防止发送过快
                    Thread.sleep(100);
                    allChannels.add(deviceChannel);
                }
            }
            if (size == 0) {
            if (allChannels.size() > 0) {
                cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag());
            }else {
                // 回复无通道
                cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), size);
                cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0);
            }
        } catch (SipException e) {
            e.printStackTrace();
        } catch (InvalidArgumentException e) {
            e.printStackTrace();
        } catch (ParseException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java

@@ -123,7 +123,7 @@

                        channelList.add(deviceChannel);
                    }
                    logger.debug("收到来自设备【{}】的通道: {}个，{}/{}", device.getDeviceId(), channelList.size(), catalogDataCatch.get(key) == null ? 0 :catalogDataCatch.get(key).size(), sumNum);
                    logger.info("收到来自设备【{}】的通道: {}个，{}/{}", device.getDeviceId(), channelList.size(), catalogDataCatch.get(key) == null ? 0 :catalogDataCatch.get(key).size(), sumNum);
                    catalogDataCatch.put(key, sumNum, device, channelList);
                    if (catalogDataCatch.get(key).size() == sumNum) {
                        // 数据已经完整接收
@@ -230,8 +230,22 @@
        }
    }

    public String getChannelSyncProgress(String deviceId) {
    public SyncStatus getChannelSyncProgress(String deviceId) {
        String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
        return catalogDataCatch.get(key) == null ? "0/0" : catalogDataCatch.get(key).size() + "/" + catalogDataCatch.getTotal(key);
        if (catalogDataCatch.get(key) == null) {
            return null;
        }else {
            return catalogDataCatch.getSyncStatus(key);
        }
    }

    public void setChannelSyncReady(String deviceId) {
        String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
        catalogDataCatch.addReady(key);
    }

    public void setChannelSyncEnd(String deviceId, String errorMsg) {
        String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
        catalogDataCatch.setChannelSyncEnd(key, errorMsg);
    }
}

 src/main/java/com/genersoft/iot/vmp/service/IDeviceService.java

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

import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;

/**
 * 设备相关业务处理
@@ -34,4 +35,24 @@
     * @return
     */
    boolean removeMobilePositionSubscribe(Device device);

    /**
     * 移除移动位置订阅
     * @param deviceId 设备ID
     * @return
     */
    SyncStatus getChannelSyncStatus(String deviceId);

    /**
     * 设置通道同步状态
     * @param deviceId 设备ID
     */
    void setChannelSyncReady(String deviceId);

    /**
     * 设置同步结束
     * @param deviceId 设备ID
     * @param errorMsg 错误信息
     */
    void setChannelSyncEnd(String deviceId, String errorMsg);
}

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

@@ -3,9 +3,12 @@
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -24,6 +27,12 @@

    @Autowired
    private ISIPCommander sipCommander;

    @Autowired
    private CatalogResponseMessageHandler catalogResponseMessageHandler;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Override
    public boolean addCatalogSubscribe(Device device) {
@@ -86,4 +95,19 @@
        dynamicTask.stop(device.getDeviceId() + "mobile_position");
        return true;
    }

    @Override
    public SyncStatus getChannelSyncStatus(String deviceId) {
        return catalogResponseMessageHandler.getChannelSyncProgress(deviceId);
    }

    @Override
    public void setChannelSyncReady(String deviceId) {
        catalogResponseMessageHandler.setChannelSyncReady(deviceId);
    }

    @Override
    public void setChannelSyncEnd(String deviceId, String errorMsg) {
        catalogResponseMessageHandler.setChannelSyncEnd(deviceId, errorMsg);
    }
}

 src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -216,4 +216,5 @@
    void sendMobilePositionMsg(JSONObject jsonObject);

    void sendStreamPushRequestedMsg(MessageForPushChannel messageForPushChannel);

}

 src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -638,4 +638,5 @@
        logger.info("[redis 推流被请求通知] {}: {}-{}", key, msg.getApp(), msg.getStream());
        redis.convertAndSend(key, (JSONObject)JSON.toJSON(msg));
    }

}

 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java

@@ -445,8 +445,6 @@
        device.setOnline(1);
        logger.info("更新设备在线: " + deviceId);
        redisCatchStorage.updateDevice(device);
        List<DeviceChannel> deviceChannelList = deviceChannelMapper.queryOnlineChannelsByDeviceId(deviceId);
        eventPublisher.catalogEventPublish(null, deviceChannelList, CatalogEvent.ON);
        return deviceMapper.update(device) > 0;
    }


 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java

@@ -4,6 +4,7 @@
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
@@ -18,6 +19,7 @@
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.kxml2.wap.wv.WV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -149,49 +151,30 @@
            @ApiImplicitParam(name="deviceId", value = "设备id", required = true, dataTypeClass = String.class),
    })
    @PostMapping("/devices/{deviceId}/sync")
    public DeferredResult<ResponseEntity<Device>> devicesSync(@PathVariable String deviceId){
    public WVPResult<SyncStatus> devicesSync(@PathVariable String deviceId){
        
        if (logger.isDebugEnabled()) {
            logger.debug("设备通道信息同步API调用，deviceId：" + deviceId);
        }
        Device device = storager.queryVideoDevice(deviceId);
        String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
        String uuid = UUID.randomUUID().toString();
        // 默认超时时间为30分钟
        DeferredResult<ResponseEntity<Device>> result = new DeferredResult<ResponseEntity<Device>>(30*60*1000L);
        result.onTimeout(()->{
            logger.warn("设备[{}]通道信息同步超时", deviceId);
            // 释放rtpserver
            RequestMessage msg = new RequestMessage();
            msg.setKey(key);
            msg.setId(uuid);
            WVPResult<Object> wvpResult = new WVPResult<>();
            wvpResult.setCode(-1);
            wvpResult.setData(device);
            wvpResult.setMsg("更新超时");
            msg.setData(wvpResult);
            resultHolder.invokeAllResult(msg);

        });
        // 等待其他相同请求返回时一起返回
        if (resultHolder.exist(key, null)) {
            resultHolder.put(key, uuid, result);
            return result;
        }else {
            cmder.catalogQuery(device, event -> {
                RequestMessage msg = new RequestMessage();
                msg.setKey(key);
                msg.setId(uuid);
                WVPResult<Object> wvpResult = new WVPResult<>();
                wvpResult.setCode(-1);
                wvpResult.setData(device);
                wvpResult.setMsg(String.format("同步通道失败，错误码： %s, %s", event.statusCode, event.msg));
                msg.setData(wvpResult);
                resultHolder.invokeAllResult(msg);
            });
            resultHolder.put(key, uuid, result);
            return result;
        SyncStatus syncStatus = deviceService.getChannelSyncStatus(deviceId);
        // 已存在则返回进度
        if (syncStatus != null && syncStatus.getErrorMsg() == null) {
            WVPResult<SyncStatus> wvpResult = new WVPResult<>();
            wvpResult.setCode(0);
            wvpResult.setData(syncStatus);
            return wvpResult;
        }
        SyncStatus syncStatusReady = new SyncStatus();
        deviceService.setChannelSyncReady(deviceId);
        cmder.catalogQuery(device, event -> {
            String errorMsg = String.format("同步通道失败，错误码： %s, %s", event.statusCode, event.msg);
            deviceService.setChannelSyncEnd(deviceId, errorMsg);
        });
        WVPResult<SyncStatus> wvpResult = new WVPResult<>();
        wvpResult.setCode(0);
        wvpResult.setMsg("开始同步");
        return wvpResult;
    }

    /**
@@ -468,4 +451,22 @@
    public WVPResult<List<DeviceChannelTree>> tree(@PathVariable String deviceId) {
        return WVPResult.Data(storager.tree(deviceId));
    }

    @GetMapping("/{deviceId}/sync_status")
    @ApiOperation(value = "获取通道同步进度", notes = "获取通道同步进度")
    public WVPResult<SyncStatus> getSyncStatus(@PathVariable String deviceId) {
        SyncStatus channelSyncStatus = deviceService.getChannelSyncStatus(deviceId);
        WVPResult<SyncStatus> wvpResult = new WVPResult<>();
        if (channelSyncStatus == null) {
            wvpResult.setCode(-1);
            wvpResult.setMsg("同步尚未开始");
        }else {
            wvpResult.setCode(0);
            wvpResult.setData(channelSyncStatus);
            if (channelSyncStatus.getErrorMsg() != null) {
                wvpResult.setMsg(channelSyncStatus.getErrorMsg());
            }
        }
        return wvpResult;
    }
}

 web_src/src/components/DeviceList.vue

@@ -57,7 +57,7 @@

                    <el-table-column label="操作" width="450" align="center" fixed="right">
                        <template slot-scope="scope">
                            <el-button size="mini" :loading="syncDevices.includes(scope.row.deviceId)"  v-if="scope.row.online!=0" icon="el-icon-refresh"  @click="refDevice(scope.row)">刷新</el-button>
              <el-button size="mini" v-if="scope.row.online!=0" icon="el-icon-refresh" @click="refDevice(scope.row)" @mouseover="getTooltipContent(scope.row.deviceId)">刷新</el-button>
                            <el-button-group>
                <el-button size="mini" icon="el-icon-video-camera-solid" v-bind:disabled="scope.row.online==0"  type="primary" @click="showChannelList(scope.row)">通道</el-button>
                <el-button size="mini" icon="el-icon-location" v-bind:disabled="scope.row.online==0"  type="primary" @click="showDevicePosition(scope.row)">定位</el-button>
@@ -78,6 +78,7 @@
                    :total="total">
                </el-pagination>
        <deviceEdit ref="deviceEdit" ></deviceEdit>
        <syncChannelProgress ref="syncChannelProgress" ></syncChannelProgress>
            </el-main>
        </el-container>
    </div>
@@ -86,11 +87,13 @@
<script>
    import uiHeader from './UiHeader.vue'
    import deviceEdit from './dialog/deviceEdit.vue'
    import syncChannelProgress from './dialog/SyncChannelProgress.vue'
    export default {
        name: 'app',
        components: {
            uiHeader,
      deviceEdit
      deviceEdit,
      syncChannelProgress,
        },
        data() {
            return {
@@ -105,7 +108,6 @@
                count:15,
                total:0,
                getDeviceListLoading: false,
        syncDevices:[]
            };
        },
        computed: {
@@ -198,8 +200,7 @@
            //刷新设备信息
            refDevice: function(itemData) {
                console.log("刷新对应设备:" + itemData.deviceId);
                var that = this;
        this.syncDevices.push(itemData.deviceId)
                let that = this;
                this.$axios({
                    method: 'post',
                    url: '/api/device/query/devices/' + itemData.deviceId + '/sync'
@@ -212,14 +213,14 @@
                            type: 'error'
                        });
                    }else{
                        that.$message({
                            showClose: true,
                            message: res.data.msg,
                            type: 'success'
                        });
                        // that.$message({
                        //     showClose: true,
                        //     message: res.data.msg,
                        //     type: 'success'
                        // });
            this.$refs.syncChannelProgress.openDialog(itemData.deviceId)
                    }
                    that.initData()
          this.syncDevices.splice(this.syncDevices.indexOf(itemData.deviceId, 1));
                }).catch((e) => {
                    console.error(e)
          that.$message({
@@ -227,8 +228,28 @@
            message: e,
            type: 'error'
          });
          this.syncDevices.splice(this.syncDevices.indexOf(itemData.deviceId, 1));
                });

            },

      getTooltipContent: async function (deviceId){
         let result = "";
         await this.$axios({
            method: 'get',
            async: false,
            url:`/api/device/query/${deviceId}/sync_status/`,
          }).then((res) => {
           if (res.data.code == 0) {
             if (res.data.data.errorMsg !== null) {
               result = res.data.data.errorMsg
             } else if (res.data.msg !== null) {
               result = res.data.msg
             } else {
               result = `同步中...[${res.data.data.current}/${res.data.data.total}]`;
             }
           }
         })
         return result;
            },
            //通知设备上传媒体流
            sendDevicePush: function(itemData) {

 web_src/src/components/dialog/SyncChannelProgress.vue

New file
@@ -0,0 +1,102 @@
<template>
  <div id="SyncChannelProgress" v-loading="isLoging">
    <el-dialog
      width="240px"
      top="13%"
      :append-to-body="true"
      :close-on-click-modal="false"
      :visible.sync="showDialog"
      :destroy-on-close="true"
      :show-close="true"
      @close="close()"
     style="text-align: center">
      <el-progress type="circle" :percentage="percentage" :status="syncStatus"></el-progress>
      <div style="text-align: center">
        {{msg}}
      </div>
    </el-dialog>
  </div>
</template>

<script>

export default {
  name: "SyncChannelProgress",
  computed: {},
  props: ['platformId'],
  created() {},
  data() {
    return {
      syncStatus: null,
      percentage: 0,
      total: 0,
      current: 0,
      showDialog: false,
      isLoging: false,
      syncFlag: false,
      deviceId: null,
      timmer: null,
      msg: "正在同步",
    };
  },
  methods: {
    openDialog: function (deviceId) {
      console.log("deviceId: " + deviceId)
      this.deviceId = deviceId;
      this.showDialog = true;
      this.msg = "";
      this.percentage= 0;
      this.total= 0;
      this.current= 0;
      this.syncFlag= false;
      this.syncStatus = null;
      this.getProgress()
    },
    getProgress(){
      this.$axios({
        method: 'get',
        url:`/api/device/query/${this.deviceId}/sync_status/`,
      }).then((res) => {
        if (res.data.code == 0) {
          if (!this.syncFlag) {
            this.syncFlag = true;
          }
          if (res.data.data == null) {
            this.syncStatus = "success"
            this.percentage = 100;
            this.msg = '同步成功';
          }else if (res.data.data.total == 0){
            this.msg = `等待同步中`;
            this.timmer = setTimeout(this.getProgress, 300)
          }else if (res.data.data.errorMsg !== null ){
            this.msg = res.data.data.errorMsg;
            this.syncStatus = "exception"
          }else {
            this.total = res.data.data.total;
            this.current = res.data.data.current;
            this.percentage = Math.floor(Number(res.data.data.current)/Number(res.data.data.total)* 10000)/100;
            this.msg = `同步中...[${res.data.data.current}/${res.data.data.total}]`;
            this.timmer = setTimeout(this.getProgress, 300)
          }
        }else {
          if (this.syncFlag) {
            this.syncStatus = "success"
            this.percentage = 100;
            this.msg = '同步成功';
          }else {
            this.syncStatus = "error"
            this.msg = res.data.msg;
          }
        }
      }).catch((error) =>{
        console.log(error);
        this.syncStatus = "error"
        this.msg = error.response.data.msg;
      });
    },
    close: function (){
      window.clearTimeout(this.timmer)
    }
  },
};
</script>