支持国标级联的目录订阅功能

648540858

2022-01-14 ac1a4a027a7bd88efb32e9da666bdba4b5fa166f

支持国标级联的目录订阅功能

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

@@ -18,13 +18,13 @@
        this.fromTag = fromHeader.getTag();
        ExpiresHeader expiresHeader = (ExpiresHeader)request.getHeader(ExpiresHeader.NAME);
        this.expires = expiresHeader.getExpires();
        this.event = (EventHeader)request.getHeader(EventHeader.NAME);
        this.event = ((EventHeader)request.getHeader(EventHeader.NAME)).getName();
    }

    private String id;
    private int expires;
    private String callId;
    private EventHeader event;
    private String event;
    private String fromTag;
    private String toTag;

@@ -38,10 +38,6 @@

    public String getCallId() {
        return callId;
    }

    public EventHeader getEvent() {
        return event;
    }

    public String getFromTag() {
@@ -68,11 +64,15 @@
        this.callId = callId;
    }

    public void setEvent(EventHeader event) {
        this.event = event;
    }

    public void setFromTag(String fromTag) {
        this.fromTag = fromTag;
    }

    public String getEvent() {
        return event;
    }

    public void setEvent(String event) {
        this.event = event;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java

@@ -1,18 +1,27 @@
package com.genersoft.iot.vmp.gb28181.event;



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

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;

import com.genersoft.iot.vmp.gb28181.bean.GbStream;

import com.genersoft.iot.vmp.gb28181.event.offline.OfflineEvent;

import com.genersoft.iot.vmp.gb28181.event.platformKeepaliveExpire.PlatformKeepaliveExpireEvent;

import com.genersoft.iot.vmp.gb28181.event.platformNotRegister.PlatformNotRegisterEvent;

import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;

import com.genersoft.iot.vmp.media.zlm.event.ZLMOfflineEvent;

import com.genersoft.iot.vmp.media.zlm.event.ZLMOnlineEvent;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.ApplicationEventPublisher;

import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Component;



import com.genersoft.iot.vmp.gb28181.bean.DeviceAlarm;

import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEvent;

import com.genersoft.iot.vmp.gb28181.event.online.OnlineEvent;



import java.util.ArrayList;

import java.util.HashSet;

import java.util.List;

import java.util.Set;



/**    

 * @description:Event事件通知推送器，支持推送在线事件、离线事件

@@ -80,4 +89,49 @@
        outEvent.setMediaServerId(mediaServerId);

        applicationEventPublisher.publishEvent(outEvent);

    }



    @Async

    public void catalogEventPublish(String platformId, DeviceChannel deviceChannel, String type) {

        List<DeviceChannel> deviceChannelList = new ArrayList<>();

        deviceChannelList.add(deviceChannel);

        catalogEventPublish(platformId, deviceChannelList, type);

    }



    @Async

    public void catalogEventPublish(String platformId, List<DeviceChannel> deviceChannels, String type) {

        CatalogEvent outEvent = new CatalogEvent(this);

        List<DeviceChannel> channels = new ArrayList<>();

        if (deviceChannels.size() > 1) {

            // 数据去重

            Set<String> gbIdSet = new HashSet<>();

            for (DeviceChannel deviceChannel : deviceChannels) {

                if (!gbIdSet.contains(deviceChannel.getChannelId())) {

                    gbIdSet.add(deviceChannel.getChannelId());

                    channels.add(deviceChannel);

                }

            }

        }else {

            channels = deviceChannels;

        }

        outEvent.setDeviceChannels(channels);

        outEvent.setType(type);

        outEvent.setPlatformId(platformId);

        applicationEventPublisher.publishEvent(outEvent);

    }



    @Async

    public void catalogEventPublishForStream(String platformId, List<GbStream> gbStreams, String type) {

        CatalogEvent outEvent = new CatalogEvent(this);

        outEvent.setGbStreams(gbStreams);

        outEvent.setType(type);

        outEvent.setPlatformId(platformId);

        applicationEventPublisher.publishEvent(outEvent);

    }



    @Async

    public void catalogEventPublishForStream(String platformId, GbStream gbStream, String type) {

        List<GbStream> gbStreamList = new ArrayList<>();

        gbStreamList.add(gbStream);

        catalogEventPublishForStream(platformId, gbStreamList, type);

    }

}


 src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepaliveTimeoutListenerForPlatform.java

@@ -2,6 +2,7 @@

import com.genersoft.iot.vmp.conf.RedisKeyExpirationEventMessageListener;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -35,6 +36,9 @@
    @Autowired
    private UserSetup userSetup;

    @Autowired
    private SipSubscribe sipSubscribe;

    public KeepaliveTimeoutListenerForPlatform(RedisMessageListenerContainer listenerContainer, UserSetup userSetup) {
        super(listenerContainer, userSetup);
    }
@@ -54,6 +58,7 @@
        String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.PLATFORM_KEEPALIVE_PREFIX + userSetup.getServerId() + "_";
        String PLATFORM_REGISTER_PREFIX = VideoManagerConstants.PLATFORM_REGISTER_PREFIX + userSetup.getServerId() + "_";
        String KEEPLIVEKEY_PREFIX = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_";
        String REGISTER_INFO_PREFIX = VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetup.getServerId() + "_";
        if (expiredKey.startsWith(PLATFORM_KEEPLIVEKEY_PREFIX)) {
            String platformGBId = expiredKey.substring(PLATFORM_KEEPLIVEKEY_PREFIX.length(),expiredKey.length());

@@ -65,6 +70,13 @@
        }else if (expiredKey.startsWith(KEEPLIVEKEY_PREFIX)){
            String deviceId = expiredKey.substring(KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
            publisher.outlineEventPublish(deviceId, KEEPLIVEKEY_PREFIX);
        }else if (expiredKey.startsWith(REGISTER_INFO_PREFIX)) {
            String callid = expiredKey.substring(REGISTER_INFO_PREFIX.length());
            SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
            eventResult.callId = callid;
            eventResult.msg = "注册超时";
            eventResult.type = "register timeout";
            sipSubscribe.getErrorSubscribe(callid).response(eventResult);
        }

    }

 src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/OfflineEventListener.java

@@ -1,6 +1,9 @@
package com.genersoft.iot.vmp.gb28181.event.offline;



import com.genersoft.iot.vmp.conf.UserSetup;

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;

import com.genersoft.iot.vmp.gb28181.event.EventPublisher;

import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

@@ -12,6 +15,8 @@
import com.genersoft.iot.vmp.common.VideoManagerConstants;

import com.genersoft.iot.vmp.storager.IVideoManagerStorager;

import com.genersoft.iot.vmp.utils.redis.RedisUtil;



import java.util.List;



/**

 * @description: 离线事件监听器，监听到离线后，修改设备离在线状态。 设备离线有两个来源：

@@ -33,6 +38,9 @@


    @Autowired

    private UserSetup userSetup;



    @Autowired

    private EventPublisher eventPublisher;



    @Override

    public void onApplicationEvent(OfflineEvent event) {

@@ -58,6 +66,8 @@
                }

        }



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

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

        // 处理离线监听

        storager.outline(event.getDeviceId());




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

@@ -3,6 +3,10 @@
import com.genersoft.iot.vmp.conf.SipConfig;

import com.genersoft.iot.vmp.conf.UserSetup;

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

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;

import com.genersoft.iot.vmp.gb28181.event.EventPublisher;

import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;

import com.genersoft.iot.vmp.service.IDeviceService;

import com.genersoft.iot.vmp.storager.dao.dto.User;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

@@ -15,6 +19,7 @@
import com.genersoft.iot.vmp.utils.redis.RedisUtil;



import java.text.SimpleDateFormat;

import java.util.List;



/**

 * @description: 在线事件监听器，监听到离线后，修改设备离在线状态。 设备在线有两个来源：

@@ -39,6 +44,9 @@


    @Autowired

    private UserSetup userSetup;



    @Autowired

    private EventPublisher eventPublisher;



    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");



@@ -76,6 +84,11 @@
        }



        device.setOnline(1);

        Device deviceInstore = storager.queryVideoDevice(device.getDeviceId());

        if (deviceInstore.getOnline() == 0) {

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

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

        }

        // 处理上线监听

        storager.updateDevice(device);




 src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEvent.java

New file
@@ -0,0 +1,58 @@
package com.genersoft.iot.vmp.gb28181.event.subscribe.catalog;

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import org.springframework.context.ApplicationEvent;

import java.util.List;

public class CatalogEvent  extends ApplicationEvent {
    public CatalogEvent(Object source) {
        super(source);
    }

    public static final String ON = "ON";         // 上线
    public static final String OFF = "OFF";       // 离线
    public static final String VLOST = "VLOST";   // 视频丢失
    public static final String DEFECT = "DEFECT"; // 故障
    public static final String ADD = "ADD";       // 增加
    public static final String DEL = "DEL";       // 删除
    public static final String UPDATE = "UPDATE";       // 更新

    private List<DeviceChannel> deviceChannels;
    private List<GbStream> gbStreams;
    private String type;
    private String platformId;

    public List<DeviceChannel> getDeviceChannels() {
        return deviceChannels;
    }

    public void setDeviceChannels(List<DeviceChannel> deviceChannels) {
        this.deviceChannels = deviceChannels;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public String getPlatformId() {
        return platformId;
    }

    public void setPlatformId(String platformId) {
        this.platformId = platformId;
    }

    public List<GbStream> getGbStreams() {
        return gbStreams;
    }

    public void setGbStreams(List<GbStream> gbStreams) {
        this.gbStreams = gbStreams;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java

New file
@@ -0,0 +1,167 @@
package com.genersoft.iot.vmp.gb28181.event.subscribe.catalog;

import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.event.platformNotRegister.PlatformNotRegisterEvent;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * catalog事件
 */
@Component
public class CatalogEventLister implements ApplicationListener<CatalogEvent> {

    private final static Logger logger = LoggerFactory.getLogger(CatalogEventLister.class);

    @Autowired
    private IVideoManagerStorager storager;
    @Autowired
    private IRedisCatchStorage redisCatchStorage;
    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private SIPCommanderFroPlatform sipCommanderFroPlatform;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Autowired
    private SipConfig config;

    @Autowired
    private UserSetup userSetup;

    @Autowired
    private IGbStreamService gbStreamService;

    @Override
    public void onApplicationEvent(CatalogEvent event) {
        SubscribeInfo subscribe = null;
        ParentPlatform parentPlatform = null;

        Map<String, List<ParentPlatform>> parentPlatformMap = new HashMap<>();
        if (event.getPlatformId() != null) {
            parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformId());
            String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() +  "_Catalog_" + event.getPlatformId();
            subscribe = redisCatchStorage.getSubscribe(key);
        }else {
            // 获取所用订阅
            List<String> platforms = redisCatchStorage.getAllSubscribePlatform();
            if (event.getDeviceChannels() != null) {
                if (platforms.size() > 0) {
                    for (DeviceChannel deviceChannel : event.getDeviceChannels()) {
                        List<ParentPlatform> parentPlatformsForGB = storager.queryPlatFormListForGBWithGBId(deviceChannel.getChannelId(), platforms);
                        parentPlatformMap.put(deviceChannel.getChannelId(), parentPlatformsForGB);
                    }
                }
            }else if (event.getGbStreams() != null) {
                if (platforms.size() > 0) {
                    for (GbStream gbStream : event.getGbStreams()) {
                        List<ParentPlatform> parentPlatformsForGB = storager.queryPlatFormListForStreamWithGBId(gbStream.getApp(),gbStream.getStream(), platforms);
                        parentPlatformMap.put(gbStream.getGbId(), parentPlatformsForGB);
                    }
                }
            }

        }
        switch (event.getType()) {
            case CatalogEvent.ON:
            case CatalogEvent.OFF:
            case CatalogEvent.DEL:

                if (parentPlatform != null || subscribe != null) {
                    List<DeviceChannel> deviceChannelList = new ArrayList<>();
                    if (event.getDeviceChannels() != null) {
                        deviceChannelList.addAll(event.getDeviceChannels());
                    }
                    if (event.getGbStreams().size() > 0){
                        for (GbStream gbStream : event.getGbStreams()) {
                            DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform.getDeviceGBId());
                            deviceChannelList.add(deviceChannelByStream);
                        }
                    }
                    if (deviceChannelList.size() > 0) {
                        logger.info("[Catalog事件: {}]平台：{}，影响通道{}个", event.getType(), event.getPlatformId(), deviceChannelList.size());
                        sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), parentPlatform, deviceChannelList, subscribe);
                    }
                }else if (parentPlatformMap.keySet().size() > 0) {
                    for (String gbId : parentPlatformMap.keySet()) {
                        List<ParentPlatform> parentPlatforms = parentPlatformMap.get(gbId);
                        if (parentPlatforms != null && parentPlatforms.size() > 0) {
                            for (ParentPlatform platform : parentPlatforms) {
                                logger.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getType(), platform.getServerGBId(), gbId);
                                String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() +  "_Catalog_" + platform.getServerGBId();
                                SubscribeInfo subscribeInfo = redisCatchStorage.getSubscribe(key);
                                List<DeviceChannel> deviceChannelList = new ArrayList<>();
                                DeviceChannel deviceChannel = new DeviceChannel();
                                deviceChannel.setChannelId(gbId);
                                deviceChannelList.add(deviceChannel);
                                sipCommanderFroPlatform.sendNotifyForCatalogOther(event.getType(), platform, deviceChannelList, subscribeInfo);
                            }
                        }
                    }
                }
                break;
            case CatalogEvent.VLOST:
                break;
            case CatalogEvent.DEFECT:
                break;
            case CatalogEvent.ADD:
            case CatalogEvent.UPDATE:
                if (parentPlatform != null || subscribe != null) {
                     List<DeviceChannel> deviceChannelList = new ArrayList<>();
                     if (event.getDeviceChannels() != null) {
                         deviceChannelList.addAll(event.getDeviceChannels());
                     }
                    if (event.getGbStreams().size() > 0){
                        for (GbStream gbStream : event.getGbStreams()) {
                            DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform.getDeviceGBId());
                            deviceChannelList.add(deviceChannelByStream);
                        }
                    }
                    if (deviceChannelList.size() > 0) {
                        logger.info("[Catalog事件: {}]平台：{}，影响通道{}个", event.getType(), event.getPlatformId(), deviceChannelList.size());
                        sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), parentPlatform, deviceChannelList, subscribe);
                    }
                }else if (parentPlatformMap.keySet().size() > 0) {
                    for (String gbId : parentPlatformMap.keySet()) {
                        List<ParentPlatform> parentPlatforms = parentPlatformMap.get(gbId);
                        if (parentPlatforms != null && parentPlatforms.size() > 0) {
                            for (ParentPlatform platform : parentPlatforms) {
                                logger.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getType(), platform.getServerGBId(), gbId);
                                String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_Catalog_" + platform.getServerGBId();
                                SubscribeInfo subscribeInfo = redisCatchStorage.getSubscribe(key);
                                List<DeviceChannel> deviceChannelList = new ArrayList<>();
                                DeviceChannel deviceChannel = storager.queryChannelInParentPlatform(platform.getServerGBId(), gbId);
                                deviceChannelList.add(deviceChannel);
                                GbStream gbStream = storager.queryStreamInParentPlatform(platform.getServerGBId(), gbId);
                                DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), platform.getDeviceGBId());
                                deviceChannelList.add(deviceChannelByStream);
                                sipCommanderFroPlatform.sendNotifyForCatalogOther(event.getType(), platform, deviceChannelList, subscribeInfo);
                            }
                        }
                    }
                }
                break;
            default:
                break;
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/task/GPSSubscribeTask.java

@@ -48,7 +48,7 @@
                        if (gbStream.isStatus()) {
                            if (gpsMsgInfo != null) {
                                // 发送GPS消息
                                sipCommanderForPlatform.sendMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
                                sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
                            }else {
                                // 没有在redis找到新的消息就使用数据库的消息
                                gpsMsgInfo = new GPSMsgInfo();
@@ -56,7 +56,7 @@
                                gpsMsgInfo.setLat(gbStream.getLongitude());
                                gpsMsgInfo.setLng(gbStream.getLongitude());
                                // 发送GPS消息
                                sipCommanderForPlatform.sendMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
                                sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
                            }
                        }


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java

@@ -14,6 +14,7 @@
import javax.sip.*;
import javax.sip.header.CSeqHeader;
import javax.sip.header.CallIdHeader;
import javax.sip.header.Header;
import javax.sip.message.Response;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -140,6 +141,7 @@
     */
    @Override
    public void processTimeout(TimeoutEvent timeoutEvent) {
        System.out.println("processTimeout");
        if(timeoutProcessor != null) {
            timeoutProcessor.process(timeoutEvent);
        }
@@ -147,14 +149,31 @@

    @Override
    public void processIOException(IOExceptionEvent exceptionEvent) {
        System.out.println("processIOException");
    }

    @Override
    public void processTransactionTerminated(TransactionTerminatedEvent transactionTerminatedEvent) {
//        Transaction transaction = null;
//        System.out.println("processTransactionTerminated");
//        if (transactionTerminatedEvent.isServerTransaction()) {
//            transaction = transactionTerminatedEvent.getServerTransaction();
//        }else {
//            transaction = transactionTerminatedEvent.getClientTransaction();
//        }
//
//        System.out.println(transaction.getBranchId());
//        System.out.println(transaction.getState());
//        System.out.println(transaction.getRequest().getMethod());
//        CallIdHeader header = (CallIdHeader)transaction.getRequest().getHeader(CallIdHeader.NAME);
//        SipSubscribe.EventResult<TransactionTerminatedEvent> terminatedEventEventResult = new SipSubscribe.EventResult<>(transactionTerminatedEvent);

//        sipSubscribe.getErrorSubscribe(header.getCallId()).response(terminatedEventEventResult);
    }

    @Override
    public void processDialogTerminated(DialogTerminatedEvent dialogTerminatedEvent) {
        System.out.println("processDialogTerminated");
        CallIdHeader callId = dialogTerminatedEvent.getDialog().getCallId();
    }


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

@@ -7,6 +7,7 @@
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;

import javax.sip.header.WWWAuthenticateHeader;
import java.util.List;

public interface ISIPCommanderForPlatform {

@@ -70,5 +71,20 @@
     * @param subscribeInfo 订阅相关的信息
     * @return
     */
    boolean sendMobilePosition(ParentPlatform parentPlatform, GPSMsgInfo gpsMsgInfo, SubscribeInfo subscribeInfo);
    boolean sendNotifyMobilePosition(ParentPlatform parentPlatform, GPSMsgInfo gpsMsgInfo, SubscribeInfo subscribeInfo);

    /**
     * 回复catalog事件-增加/更新
     * @param parentPlatform
     * @param deviceChannels
     */
    boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo);

    /**
     * 回复catalog事件-删除
     * @param parentPlatform
     * @param deviceChannels
     */
    boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo);

}

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

@@ -23,6 +23,7 @@
import javax.sip.header.WWWAuthenticateHeader;
import javax.sip.message.Request;
import java.text.ParseException;
import java.util.List;
import java.util.UUID;

@Component
@@ -96,7 +97,8 @@

                request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(Request.REGISTER), "FromRegister" + tm, null, callIdHeader);
                // 将 callid 写入缓存， 等注册成功可以更新状态
                redisCatchStorage.updatePlatformRegisterInfo(callIdHeader.getCallId(), parentPlatform.getServerGBId());
                String callIdFromHeader = callIdHeader.getCallId();
                redisCatchStorage.updatePlatformRegisterInfo(callIdFromHeader, parentPlatform.getServerGBId());

                sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (event)->{
                    if (event != null) {
@@ -104,6 +106,7 @@
                                parentPlatform.getServerGBId(),
                                event.msg);
                    }
                    redisCatchStorage.delPlatformRegisterInfo(callIdFromHeader);
                    if (errorEvent != null ) {
                        errorEvent.response(event);
                    }
@@ -219,8 +222,8 @@
                catalogXml.append("<Owner>" + channel.getOwner() + "</Owner>\r\n");
                catalogXml.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n");
                catalogXml.append("<Address>" + channel.getAddress() + "</Address>\r\n");
                catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");// TODO 当前不能添加分组， 所以暂时没有父节点
                catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n"); // TODO 当前不能添加分组， 所以暂时没有父节点
                catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");
                catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n");
                catalogXml.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n");
                catalogXml.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n");
                catalogXml.append("<Status>" + (channel.getStatus() == 0?"OFF":"ON") + "</Status>\r\n");
@@ -329,7 +332,7 @@
    }

    @Override
    public boolean sendMobilePosition(ParentPlatform parentPlatform, GPSMsgInfo gpsMsgInfo, SubscribeInfo subscribeInfo) {
    public boolean sendNotifyMobilePosition(ParentPlatform parentPlatform, GPSMsgInfo gpsMsgInfo, SubscribeInfo subscribeInfo) {
        if (parentPlatform == null) {
            return false;
        }
@@ -364,4 +367,110 @@
        }
        return true;
    }

    @Override
    public boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo) {
        if (parentPlatform == null) {
            return false;
        }
        if (deviceChannels == null || deviceChannels.size() == 0) {
            return false;
        }
        for (DeviceChannel channel : deviceChannels) {
            try {
                StringBuffer catalogXml = new StringBuffer(600);
                catalogXml.append("<?xml version=\"1.0\" encoding=\"GB2312\"?>\r\n");
                catalogXml.append("<Notify>\r\n");
                catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
                catalogXml.append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n");
                catalogXml.append("<SumNum>" + deviceChannels.size() + "</SumNum>\r\n");
                catalogXml.append("<DeviceList Num=\"1\">\r\n");
                catalogXml.append("<Item>\r\n");
                catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
                catalogXml.append("<Event>" + type + "</Event>\r\n");
                catalogXml.append("<Name>" + channel.getName() + "</Name>\r\n");
                catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n");
                catalogXml.append("<Model>" + channel.getModel() + "</Model>\r\n");
                catalogXml.append("<Owner>" + channel.getOwner() + "</Owner>\r\n");
                catalogXml.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n");
                catalogXml.append("<Address>" + channel.getAddress() + "</Address>\r\n");
                catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");
                catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n");
                catalogXml.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n");
                catalogXml.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n");
                catalogXml.append("<Status>" + (channel.getStatus() == 0 ? "OFF" : "ON") + "</Status>\r\n");
                catalogXml.append("<Longitude>" + channel.getLongitude() + "</Longitude>\r\n");
                catalogXml.append("<Latitude>" + channel.getLatitude() + "</Latitude>\r\n");
                catalogXml.append("<IPAddress>" + channel.getIpAddress() + "</IPAddress>\r\n");
                catalogXml.append("<Port>" + channel.getPort() + "</Port>\r\n");
                catalogXml.append("<Info>\r\n");
                catalogXml.append("<PTZType>" + channel.getPTZType() + "</PTZType>\r\n");
                catalogXml.append("</Info>\r\n");
                catalogXml.append("</Item>\r\n");
                catalogXml.append("</DeviceList>\r\n");
                catalogXml.append("</Notify>\r\n");

                CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                        : udpSipProvider.getNewCallId();
                callIdHeader.setCallId(subscribeInfo.getCallId());

                String tm = Long.toString(System.currentTimeMillis());

                Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXml.toString(), subscribeInfo.getToTag(), subscribeInfo.getFromTag(), callIdHeader);
                transmitRequest(parentPlatform, request);
                Thread.sleep(100);
            } catch (SipException | ParseException | InvalidArgumentException e) {
                e.printStackTrace();
                return false;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return true;
    }

    @Override
    public boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo) {
        if (parentPlatform == null) {
            return false;
        }
        if (deviceChannels == null || deviceChannels.size() == 0) {
            return false;
        }

        for (DeviceChannel channel : deviceChannels) {
            try {
                StringBuffer catalogXml = new StringBuffer(600);
                catalogXml.append("<?xml version=\"1.0\" encoding=\"GB2312\"?>\r\n");
                catalogXml.append("<Notify>\r\n");
                catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
                catalogXml.append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n");
                catalogXml.append("<SumNum>" + deviceChannels.size() + "</SumNum>\r\n");
                catalogXml.append("<DeviceList Num=\"1\">\r\n");
                catalogXml.append("<Item>\r\n");
                catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
                catalogXml.append("<Event>" + type + "</Event>\r\n");
                catalogXml.append("</Item>\r\n");
                catalogXml.append("</DeviceList>\r\n");
                catalogXml.append("</Notify>\r\n");

                CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                        : udpSipProvider.getNewCallId();
                callIdHeader.setCallId(subscribeInfo.getCallId());

                String tm = Long.toString(System.currentTimeMillis());

                Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXml.toString(), subscribeInfo.getToTag(), subscribeInfo.getFromTag(), callIdHeader);
                transmitRequest(parentPlatform, request);
                Thread.sleep(100);
            } catch (SipException | ParseException | InvalidArgumentException e) {
                e.printStackTrace();
                return false;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return true;
    }

}

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

@@ -106,9 +106,8 @@
            if (platform != null) {
                // 查询平台下是否有该通道
                DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
                List<GbStream> gbStreams = storager.queryStreamInParentPlatform(requesterId, channelId);
                GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
                PlatformCatalog catalog = storager.getCatalog(channelId);
                GbStream gbStream = gbStreams.size() > 0? gbStreams.get(0):null;
                MediaServerItem mediaServerItem = null;
                // 不是通道可能是直播流
                if (channel != null && gbStream == null ) {

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

@@ -6,6 +6,7 @@
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
@@ -49,6 +50,9 @@

    @Autowired
    private IVideoManagerStorager storager;

    @Autowired
    private EventPublisher eventPublisher;

    @Autowired
    private SipConfig sipConfig;
@@ -259,39 +263,39 @@
                    channel.setDeviceId(device.getDeviceId());
                    logger.debug("收到来自设备【{}】的通道: {}【{}】", device.getDeviceId(), channel.getName(), channel.getChannelId());
                    switch (eventElement.getText().toUpperCase()) {
                        case "ON" : // 上线
                        case CatalogEvent.ON: // 上线
                            logger.info("收到来自设备【{}】的通道【{}】上线通知", device.getDeviceId(), channel.getChannelId());
                            storager.deviceChannelOnline(deviceId, channel.getChannelId());
                            // 回复200 OK
                            responseAck(evt, Response.OK);
                            break;
                        case "OFF" : // 离线
                        case CatalogEvent.OFF : // 离线
                            logger.info("收到来自设备【{}】的通道【{}】离线通知", device.getDeviceId(), channel.getChannelId());
                            storager.deviceChannelOffline(deviceId, channel.getChannelId());
                            // 回复200 OK
                            responseAck(evt, Response.OK);
                            break;
                        case "VLOST" : // 视频丢失
                        case CatalogEvent.VLOST: // 视频丢失
                            logger.info("收到来自设备【{}】的通道【{}】视频丢失通知", device.getDeviceId(), channel.getChannelId());
                            storager.deviceChannelOffline(deviceId, channel.getChannelId());
                            // 回复200 OK
                            responseAck(evt, Response.OK);
                            break;
                        case "DEFECT" : // 故障
                        case CatalogEvent.DEFECT: // 故障
                            // 回复200 OK
                            responseAck(evt, Response.OK);
                            break;
                        case "ADD" : // 增加
                        case CatalogEvent.ADD: // 增加
                            logger.info("收到来自设备【{}】的增加通道【{}】通知", device.getDeviceId(), channel.getChannelId());
                            storager.updateChannel(deviceId, channel);
                            responseAck(evt, Response.OK);
                            break;
                        case "DEL" : // 删除
                        case CatalogEvent.DEL: // 删除
                            logger.info("收到来自设备【{}】的删除通道【{}】通知", device.getDeviceId(), channel.getChannelId());
                            storager.delChannel(deviceId, channel.getChannelId());
                            responseAck(evt, Response.OK);
                            break;
                        case "UPDATE" : // 更新
                        case CatalogEvent.UPDATE: // 更新
                            logger.info("收到来自设备【{}】的更新通道【{}】通知", device.getDeviceId(), channel.getChannelId());
                            storager.updateChannel(deviceId, channel);
                            responseAck(evt, Response.OK);
@@ -300,6 +304,8 @@
                            responseAck(evt, Response.BAD_REQUEST, "event not found");

                    }
                    // 转发变化信息
                    eventPublisher.catalogEventPublish(null, channel, eventElement.getText().toUpperCase());

                }


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

@@ -85,9 +85,9 @@
//            } else if (CmdType.ALARM.equals(cmd)) {
//                logger.info("接收到Alarm订阅");
//                processNotifyAlarm(evt, rootElement);
//            } else if (CmdType.CATALOG.equals(cmd)) {
//                logger.info("接收到Catalog订阅");
//                processNotifyCatalogList(evt, rootElement);
            } else if (CmdType.CATALOG.equals(cmd)) {
                logger.info("接收到Catalog订阅");
                processNotifyCatalogList(evt, rootElement);
            } else {
                logger.info("接收到消息：" + cmd);
//                responseAck(evt, Response.OK);
@@ -177,7 +177,40 @@
    }

    private void processNotifyCatalogList(RequestEvent evt, Element rootElement) {
        String platformId = SipUtils.getUserIdFromFromHeader(evt.getRequest());
        String deviceID = XmlUtil.getText(rootElement, "DeviceID");
        SubscribeInfo subscribeInfo = new SubscribeInfo(evt, platformId);
        String sn = XmlUtil.getText(rootElement, "SN");
        String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() +  "_Catalog_" + platformId;

        StringBuilder resultXml = new StringBuilder(200);
        resultXml.append("<?xml version=\"1.0\" ?>\r\n")
                .append("<Response>\r\n")
                .append("<CmdType>Catalog</CmdType>\r\n")
                .append("<SN>" + sn + "</SN>\r\n")
                .append("<DeviceID>" + deviceID + "</DeviceID>\r\n")
                .append("<Result>OK</Result>\r\n")
                .append("</Response>\r\n");

        if (subscribeInfo.getExpires() > 0) {
            redisCatchStorage.updateSubscribe(key, subscribeInfo);
        }else if (subscribeInfo.getExpires() == 0) {
            redisCatchStorage.delSubscribe(key);
        }

        try {
            Response response = responseXmlAck(evt, resultXml.toString());
            ToHeader toHeader = (ToHeader)response.getHeader(ToHeader.NAME);
            subscribeInfo.setToTag(toHeader.getTag());
            redisCatchStorage.updateSubscribe(key, subscribeInfo);

        } catch (SipException e) {
            e.printStackTrace();
        } catch (InvalidArgumentException e) {
            e.printStackTrace();
        } catch (ParseException e) {
            e.printStackTrace();
        }
    }

}

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.media.zlm;



import java.util.ArrayList;

import java.util.List;

import java.util.UUID;



@@ -8,6 +9,9 @@
import com.genersoft.iot.vmp.conf.MediaConfig;

import com.genersoft.iot.vmp.conf.UserSetup;

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

import com.genersoft.iot.vmp.gb28181.bean.GbStream;

import com.genersoft.iot.vmp.gb28181.event.EventPublisher;

import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;

import com.genersoft.iot.vmp.media.zlm.dto.*;

import com.genersoft.iot.vmp.service.*;

import com.genersoft.iot.vmp.service.bean.SSRCInfo;

@@ -65,7 +69,7 @@
    private IMediaService mediaService;



    @Autowired

    private ZLMRESTfulUtils zlmresTfulUtils;

    private EventPublisher eventPublisher;



     @Autowired

     private ZLMMediaListManager zlmMediaListManager;

@@ -341,29 +345,52 @@
                if (!"rtp".equals(app)){

                    String type = OriginType.values()[item.getOriginType()].getType();

                    MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);



                    if (mediaServerItem != null){

                        if (regist) {

                            StreamPushItem streamPushItem = null;

                            redisCatchStorage.addStream(mediaServerItem, type, app, streamId, item);

                            if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()

                                    || item.getOriginType() == OriginType.RTMP_PUSH.ordinal()

                                    || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {

                                zlmMediaListManager.addPush(item);

                                streamPushItem = zlmMediaListManager.addPush(item);

                            }

                            List<GbStream> gbStreams = new ArrayList<>();

                            if (streamPushItem == null || streamPushItem.getGbId() == null) {

                                GbStream gbStream = storager.getGbStream(app, streamId);

                                gbStreams.add(gbStream);

                            }else {

                                if (streamPushItem.getGbId() != null) {

                                    gbStreams.add(streamPushItem);

                                }

                            }

                            if (gbStreams.size() > 0) {

                                eventPublisher.catalogEventPublishForStream(null, gbStreams, CatalogEvent.ON);

                            }



                        }else {

                            // 兼容流注销时类型从redis记录获取

                            MediaItem mediaItem = redisCatchStorage.getStreamInfo(app, streamId, mediaServerId);

                            type = OriginType.values()[mediaItem.getOriginType()].getType();

                            if (mediaItem != null) {

                                type = OriginType.values()[mediaItem.getOriginType()].getType();

                                redisCatchStorage.removeStream(mediaServerItem.getId(), type, app, streamId);

                            }

                            GbStream gbStream = storager.getGbStream(app, streamId);

                            if (gbStream != null) {

                                eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);

                            }

                            zlmMediaListManager.removeMedia(app, streamId);

                            redisCatchStorage.removeStream(mediaServerItem.getId(), type, app, streamId);

                        }

                        // 发送流变化redis消息

                        JSONObject jsonObject = new JSONObject();

                        jsonObject.put("serverId", userSetup.getServerId());

                        jsonObject.put("app", app);

                        jsonObject.put("stream", streamId);

                        jsonObject.put("register", regist);

                        jsonObject.put("mediaServerId", mediaServerId);

                        redisCatchStorage.sendStreamChangeMsg(type, jsonObject);

                        if (type != null) {

                            // 发送流变化redis消息

                            JSONObject jsonObject = new JSONObject();

                            jsonObject.put("serverId", userSetup.getServerId());

                            jsonObject.put("app", app);

                            jsonObject.put("stream", streamId);

                            jsonObject.put("register", regist);

                            jsonObject.put("mediaServerId", mediaServerId);

                            redisCatchStorage.sendStreamChangeMsg(type, jsonObject);

                        }

                    }

                }

            }


 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java

@@ -105,7 +105,7 @@
        updateMedia(mediaServerItem, app, streamId);
    }

    public void addPush(MediaItem mediaItem) {
    public StreamPushItem addPush(MediaItem mediaItem) {
        // 查找此直播流是否存在redis预设gbId
        StreamPushItem transform = streamPushService.transform(mediaItem);
        // 从streamId取出查询关键值
@@ -130,7 +130,6 @@
                for (GbStream gbStream : gbStreams) {
                    // 出现使用相同国标Id的视频流时，使用新流替换旧流，
                    gbStreamMapper.del(gbStream.getApp(), gbStream.getStream());
                    platformGbStreamMapper.delByAppAndStream(gbStream.getApp(), gbStream.getStream());
                    if (!gbStream.isStatus()) {
                        streamPushMapper.del(gbStream.getApp(), gbStream.getStream());
                    }
@@ -142,6 +141,7 @@
                gbStreamMapper.add(transform);
            }
        }
        return transform;
    }



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

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

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.github.pagehelper.PageInfo;

@@ -35,6 +36,11 @@
    /**
     * 移除国标关联
     * @param gbStreams
     * @param platformId
     */
    boolean delPlatformInfo(List<GbStream> gbStreams);
    boolean delPlatformInfo(String platformId, List<GbStream> gbStreams);

    DeviceChannel getDeviceChannelListByStream(GbStream gbStream, String catalogId, String deviceGBId);

    void sendCatalogMsg(GbStream gbStream, String type);
}

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

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

import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.github.pagehelper.PageHelper;
@@ -14,6 +21,7 @@
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import java.util.ArrayList;
import java.util.List;

@Service
@@ -33,6 +41,15 @@
    @Autowired
    private PlatformGbStreamMapper platformGbStreamMapper;

    @Autowired
    private ParentPlatformMapper platformMapper;

    @Autowired
    private SipConfig sipConfig;

    @Autowired
    private EventPublisher eventPublisher;

    @Override
    public PageInfo<GbStream> getAll(Integer page, Integer count) {
        PageHelper.startPage(page, count);
@@ -51,32 +68,62 @@
        // 放在事务内执行
        boolean result = false;
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        ParentPlatform parentPlatform = platformMapper.getParentPlatByServerGBId(platformId);
        try {
            List<DeviceChannel> deviceChannelList = new ArrayList<>();
            for (GbStream gbStream : gbStreams) {
                gbStream.setCatalogId(catalogId);
                gbStream.setPlatformId(platformId);
                // TODO 修改为批量提交
                platformGbStreamMapper.add(gbStream);
                DeviceChannel deviceChannelListByStream = getDeviceChannelListByStream(gbStream, catalogId, parentPlatform.getDeviceGBId());
                deviceChannelList.add(deviceChannelListByStream);
            }
            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
            eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.ADD);
            result = true;
        }catch (Exception e) {
            logger.error("批量保存流与平台的关系时错误", e);
            dataSourceTransactionManager.rollback(transactionStatus);
        }
        return result;

    }

    @Override
    public boolean delPlatformInfo(List<GbStream> gbStreams) {
    public DeviceChannel getDeviceChannelListByStream(GbStream gbStream, String catalogId, String deviceGBId) {
        DeviceChannel deviceChannel = new DeviceChannel();
        deviceChannel.setChannelId(gbStream.getGbId());
        deviceChannel.setName(gbStream.getName());
        deviceChannel.setLongitude(gbStream.getLongitude());
        deviceChannel.setLatitude(gbStream.getLatitude());
        deviceChannel.setDeviceId(deviceGBId);
        deviceChannel.setManufacture("wvp-pro");
        deviceChannel.setStatus(gbStream.isStatus()?1:0);
        deviceChannel.setParentId(catalogId ==null?gbStream.getCatalogId():catalogId);
        deviceChannel.setRegisterWay(1);
        deviceChannel.setCivilCode(sipConfig.getDomain());
        deviceChannel.setModel("live");
        deviceChannel.setOwner("wvp-pro");
        deviceChannel.setParental(0);
        deviceChannel.setSecrecy("0");
        return deviceChannel;
    }

    @Override
    public boolean delPlatformInfo(String platformId, List<GbStream> gbStreams) {
        // 放在事务内执行
        boolean result = false;
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        try {
            List<DeviceChannel> deviceChannelList = new ArrayList<>();
            for (GbStream gbStream : gbStreams) {
                platformGbStreamMapper.delByAppAndStream(gbStream.getApp(), gbStream.getStream());
                DeviceChannel deviceChannel = new DeviceChannel();
                deviceChannel.setChannelId(gbStream.getGbId());
                deviceChannelList.add(deviceChannel);
                eventPublisher.catalogEventPublish(platformId, deviceChannel, CatalogEvent.DEL);
            }

            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
            result = true;
        }catch (Exception e) {
@@ -85,4 +132,27 @@
        }
        return result;
    }

    @Override
    public void sendCatalogMsg(GbStream gbStream, String type) {
        List<GbStream> gbStreams = new ArrayList<>();
        if (gbStream.getGbId() != null) {
            gbStreams.add(gbStream);
        }else {
            StreamProxyItem streamProxyItem = gbStreamMapper.selectOne(gbStream.getApp(), gbStream.getStream());
            if (streamProxyItem != null && streamProxyItem.getGbId() != null){
                gbStreams.add(streamProxyItem);
            }
        }
        if (gbStreams.size() > 0) {
            for (GbStream gs : gbStreams) {
                List<ParentPlatform> parentPlatforms = platformGbStreamMapper.selectByAppAndStream(gs.getApp(), gs.getStream());
                if (parentPlatforms.size() > 0) {
                    for (ParentPlatform parentPlatform : parentPlatforms) {
                        eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), gs, type);
                    }
                }
            }
        }
    }
}

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

@@ -2,9 +2,13 @@

import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
@@ -58,10 +62,16 @@
    private UserSetup userSetup;

    @Autowired
    private SipConfig sipConfig;

    @Autowired
    private GbStreamMapper gbStreamMapper;

    @Autowired
    private PlatformGbStreamMapper platformGbStreamMapper;

    @Autowired
    private EventPublisher eventPublisher;

    @Autowired
    private ParentPlatformMapper parentPlatformMapper;
@@ -146,6 +156,7 @@
                StreamProxyItem streamProxyItems = platformGbStreamMapper.selectOne(param.getApp(), stream, parentPlatform.getServerGBId());
                if (streamProxyItems == null) {
                    platformGbStreamMapper.add(param);
                    eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), param, CatalogEvent.ADD);
                }
            }
        }
@@ -194,6 +205,7 @@
    public void del(String app, String stream) {
        StreamProxyItem streamProxyItem = videoManagerStorager.queryStreamProxy(app, stream);
        if (streamProxyItem != null) {
            gbStreamService.sendCatalogMsg(streamProxyItem, CatalogEvent.DEL);
            videoManagerStorager.deleteStreamProxy(app, stream);
            JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
            if (jsonObject != null && jsonObject.getInteger("code") == 0) {

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

@@ -5,12 +5,16 @@
import com.alibaba.fastjson.TypeReference;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@@ -40,6 +44,12 @@

    @Autowired
    private PlatformGbStreamMapper platformGbStreamMapper;

    @Autowired
    private IGbStreamService gbStreamService;

    @Autowired
    private EventPublisher eventPublisher;

    @Autowired
    private ZLMRESTfulUtils zlmresTfulUtils;
@@ -115,6 +125,7 @@
        stream.setStreamType("push");
        stream.setStatus(true);
        int add = gbStreamMapper.add(stream);

        // 查找开启了全部直播流共享的上级平台
        List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream();
        if (parentPlatforms.size() > 0) {
@@ -122,18 +133,30 @@
                stream.setCatalogId(parentPlatform.getCatalogId());
                stream.setPlatformId(parentPlatform.getServerGBId());
                String streamId = stream.getStream();
                StreamProxyItem streamProxyItems = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId());
                if (streamProxyItems == null) {
                StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId());
                if (streamProxyItem == null) {
                    platformGbStreamMapper.add(stream);
                    eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD);
                }else {
                    if (!streamProxyItem.getGbId().equals(stream.getGbId())) {
                        // 此流使用另一个国标Id已经与该平台关联，移除此记录
                        platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId());
                        platformGbStreamMapper.add(stream);
                        eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD);
                    }
                }
            }
        }

        return add > 0;
    }

    @Override
    public boolean removeFromGB(GbStream stream) {
        // 判断是否需要发送事件
        gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL);
        int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
        platformGbStreamMapper.delByAppAndStream(stream.getApp(), stream.getStream());
        MediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
        JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, stream.getApp(), stream.getStream());
        if (mediaList == null) {
@@ -152,6 +175,8 @@
    @Override
    public boolean stop(String app, String streamId) {
        StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
        gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);

        int delStream = streamPushMapper.del(app, streamId);
        gbStreamMapper.del(app, streamId);
        platformGbStreamMapper.delByAppAndStream(app, streamId);

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

@@ -210,4 +210,8 @@
    void delSubscribe(String key);

    MediaItem getStreamInfo(String app, String streamId, String mediaServerId);

    List<SubscribeInfo> getAllSubscribe();

    List<String> getAllSubscribePlatform();
}

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

@@ -100,6 +100,7 @@
     * @return

     */

    public List<DeviceChannel> queryChannelsByDeviceId(String deviceId);

    public List<DeviceChannel> queryOnlineChannelsByDeviceId(String deviceId);



    /**

     * 获取某个设备的通道

@@ -341,7 +342,7 @@
     * @param channelId

     * @return

     */

    List<GbStream> queryStreamInParentPlatform(String platformId, String channelId);

    GbStream queryStreamInParentPlatform(String platformId, String channelId);



    /**

     * 获取平台关联的直播流

@@ -459,4 +460,10 @@
    int delRelation(PlatformCatalog platformCatalog);



    int updateStreamGPS(List<GPSMsgInfo> gpsMsgInfo);



    List<ParentPlatform> queryPlatFormListForGBWithGBId(String channelId, List<String> platforms);



    List<ParentPlatform> queryPlatFormListForStreamWithGBId(String app, String stream, List<String> platforms);



    GbStream getGbStream(String app, String streamId);

}


 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java

@@ -54,17 +54,22 @@
    int update(DeviceChannel channel);

    @Select(value = {" <script>" +
            "SELECT * FROM ( "+
            " SELECT * , (SELECT count(0) FROM device_channel WHERE parentId=dc.channelId) as subCount FROM device_channel dc " +
            " WHERE dc.deviceId=#{deviceId} " +
            " <if test='query != null'> AND (dc.channelId LIKE '%${query}%' OR dc.name LIKE '%${query}%' OR dc.name LIKE '%${query}%')</if> " +
            " <if test='parentChannelId != null'> AND dc.parentId=#{parentChannelId} </if> " +
            " <if test='online == true' > AND dc.status=1</if>" +
            " <if test='online == false' > AND dc.status=0</if>) dcr" +
            " WHERE 1=1 " +
            "SELECT " +
            "dc1.*, " +
            "COUNT(dc2.channelId) as subCount " +
            "from " +
            "device_channel dc1 " +
            "left join device_channel dc2 on " +
            "dc1.channelId = dc2.parentId " +
            "WHERE " +
            "dc1.deviceId = #{deviceId} " +
            " <if test='query != null'> AND (dc1.channelId LIKE '%${query}%' OR dc1.name LIKE '%${query}%' OR dc1.name LIKE '%${query}%')</if> " +
            " <if test='parentChannelId != null'> AND dc1.parentId=#{parentChannelId} </if> " +
            " <if test='online == true' > AND dc1.status=1</if>" +
            " <if test='online == false' > AND dc1.status=0</if>" +
            " <if test='hasSubChannel == true' >  AND subCount >0</if>" +
            " <if test='hasSubChannel == false' >  AND subCount=0</if>" +
            " ORDER BY channelId ASC" +
            "GROUP BY dc1.channelId " +
            " </script>"})
    List<DeviceChannel> queryChannels(String deviceId, String parentChannelId, String query, Boolean hasSubChannel, Boolean online);

@@ -170,19 +175,30 @@
            "</script>"})
    int batchUpdate(List<DeviceChannel> updateChannels);


    @Select(value = {" <script>" +
            "SELECT * FROM ( "+
            " SELECT * , (SELECT count(0) FROM device_channel WHERE parentId=dc.channelId) as subCount FROM device_channel dc " +
            " WHERE dc.deviceId=#{deviceId} " +
            " <if test='query != null'> AND (dc.channelId LIKE '%${query}%' OR dc.name LIKE '%${query}%' OR dc.name LIKE '%${query}%')</if> " +
            " <if test='parentChannelId != null'> AND dc.parentId=#{parentChannelId} </if> " +
            " <if test='online == true' > AND dc.status=1</if>" +
            " <if test='online == false' > AND dc.status=0</if>) dcr" +
            " WHERE 1=1 " +
            "SELECT " +
            "dc1.*, " +
            "COUNT(dc2.channelId) as subCount " +
            "from " +
            "device_channel dc1 " +
            "left join device_channel dc2 on " +
            "dc1.channelId = dc2.parentId " +
            "WHERE " +
            "dc1.deviceId = #{deviceId} " +
            " <if test='query != null'> AND (dc1.channelId LIKE '%${query}%' OR dc1.name LIKE '%${query}%' OR dc1.name LIKE '%${query}%')</if> " +
            " <if test='parentChannelId != null'> AND dc1.parentId=#{parentChannelId} </if> " +
            " <if test='online == true' > AND dc1.status=1</if>" +
            " <if test='online == false' > AND dc1.status=0</if>" +
            " <if test='hasSubChannel == true' >  AND subCount >0</if>" +
            " <if test='hasSubChannel == false' >  AND subCount=0</if>" +
            " ORDER BY channelId ASC" +
            " LIMIT #{limit} OFFSET #{start}" +
            "GROUP BY dc1.channelId " +
            "ORDER BY dc1.channelId ASC " +
            "Limit #{limit} OFFSET #{start}" +
            " </script>"})
    List<DeviceChannel> queryChannelsByDeviceIdWithStartAndLimit(String deviceId, String parentChannelId, String query, Boolean hasSubChannel, Boolean online, int start, int limit);
    List<DeviceChannel> queryChannelsByDeviceIdWithStartAndLimit(String deviceId, String parentChannelId, String query,
                                                                 Boolean hasSubChannel, Boolean online, int start, int limit);

    @Select("SELECT * FROM device_channel WHERE deviceId=#{deviceId} AND status=1")
    List<DeviceChannel> queryOnlineChannelsByDeviceId(String deviceId);
}

 src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java

@@ -13,7 +13,7 @@
@Repository
public interface GbStreamMapper {

    @Insert("INSERT INTO gb_stream (app, stream, gbId, name, " +
    @Insert("REPLACE INTO gb_stream (app, stream, gbId, name, " +
            "longitude, latitude, streamType, mediaServerId, status) VALUES" +
            "('${app}', '${stream}', '${gbId}', '${name}', " +
            "'${longitude}', '${latitude}', '${streamType}', " +
@@ -48,7 +48,7 @@
    @Select("SELECT gs.*, pgs.platformId as platformId, pgs.catalogId as catalogId FROM gb_stream gs " +
            "LEFT JOIN platform_gb_stream pgs ON gs.app = pgs.app AND gs.stream = pgs.stream " +
            "WHERE gs.gbId = '${gbId}' AND pgs.platformId = '${platformId}'")
    List<GbStream> queryStreamInPlatform(String platformId, String gbId);
    GbStream queryStreamInPlatform(String platformId, String gbId);

    @Select("SELECT gs.*, pgs.platformId as platformId, pgs.catalogId as catalogId FROM gb_stream gs " +
            "LEFT JOIN platform_gb_stream pgs ON gs.app = pgs.app AND gs.stream = pgs.stream " +

 src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java

@@ -2,6 +2,7 @@

import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
import org.apache.ibatis.annotations.Delete;
@@ -73,4 +74,18 @@
           "DELETE FROM platform_gb_channel WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
           "</script>")
    int delByCatalogIdAndChannelIdAndPlatformId(PlatformCatalog platformCatalog);

    @Select("<script> " +
            "SELECT " +
            "pp.* " +
            "FROM " +
            "parent_platform pp " +
            "left join platform_gb_channel pgc on " +
            "pp.serverGBId = pgc.platformId " +
            "WHERE " +
            "pgc.channelId = #{channelId} " +
            "AND pp.serverGBId IN" +
            "<foreach collection='platforms'  item='item'  open='(' separator=',' close=')' > #{item}</foreach>" +
            "</script> ")
    List<ParentPlatform> queryPlatFormListForGBWithGBId(String channelId, List<String> platforms);
}

 src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformGbStreamMapper.java

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

import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
import com.genersoft.iot.vmp.gb28181.bean.PlatformGbStream;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
@@ -14,7 +15,7 @@
@Repository
public interface PlatformGbStreamMapper {

    @Insert("INSERT INTO platform_gb_stream (app, stream, platformId, catalogId) VALUES" +
    @Insert("REPLACE INTO platform_gb_stream (app, stream, platformId, catalogId) VALUES" +
            "('${app}', '${stream}', '${platformId}', '${catalogId}')")
    int add(PlatformGbStream platformGbStream);

@@ -24,10 +25,20 @@
    @Delete("DELETE FROM platform_gb_stream WHERE platformId=#{platformId}")
    int delByPlatformId(String platformId);

    @Select("SELECT * FROM platform_gb_stream WHERE app=#{app} AND stream=#{stream}")
    List<StreamProxyItem> selectByAppAndStream(String app, String stream);
    @Select("SELECT " +
            "pp.* " +
            "FROM " +
            "platform_gb_stream pgs " +
            "LEFT JOIN parent_platform pp ON pp.serverGBId = pgs.platformId " +
            "WHERE " +
            "pgs.app =#{app} " +
            "AND pgs.stream =#{stream} " +
            "GROUP BY pp.serverGBId")
    List<ParentPlatform> selectByAppAndStream(String app, String stream);

    @Select("SELECT * FROM platform_gb_stream WHERE app=#{app} AND stream=#{stream} AND platformId=#{serverGBId}")
    @Select("SELECT pgs.*, gs.gbId  FROM platform_gb_stream pgs " +
            "LEFT JOIN gb_stream gs ON pgs.app = gs.app AND pgs.stream = gs.stream " +
            "WHERE pgs.app=#{app} AND pgs.stream=#{stream} AND pgs.platformId=#{serverGBId}")
    StreamProxyItem selectOne(String app, String stream, String serverGBId);

    @Select("select gs.* \n" +
@@ -47,4 +58,21 @@
    @Delete("DELETE FROM platform_gb_stream WHERE catalogId=#{id}")
    int delByCatalogId(String id);

    @Select("<script> " +
            "SELECT " +
            "pp.* " +
            "FROM " +
            "parent_platform pp " +
            "left join platform_gb_stream pgs on " +
            "pp.serverGBId = pgs.platformId " +
            "WHERE " +
            "pgs.app = #{app} " +
            "AND pgs.stream = #{stream}" +
            "AND pp.serverGBId IN" +
            "<foreach collection='platforms'  item='item'  open='(' separator=',' close=')' > #{item}</foreach>" +
            "</script> ")
    List<ParentPlatform> queryPlatFormListForGBWithGBId(String app, String stream, List<String> platforms);

    @Select("SELECT * FROM platform_gb_stream WHERE app=#{app} AND stream=#{stream} AND platformId=#{platformId}")
    int delByAppAndStreamAndPlatform(String app, String streamId, String platformId);
}

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

@@ -250,7 +250,7 @@
    @Override
    public void updatePlatformRegisterInfo(String callId, String platformGbId) {
        String key = VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + userSetup.getServerId() + "_" + callId;
        redis.set(key, platformGbId);
        redis.set(key, platformGbId, 30);
    }


@@ -508,4 +508,30 @@

        return result;
    }

    @Override
    public List<SubscribeInfo> getAllSubscribe() {
        String scanKey = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() +  "_Catalog_*";
        List<SubscribeInfo> result = new ArrayList<>();
        List<Object> keys = redis.scan(scanKey);
        for (int i = 0; i < keys.size(); i++) {
            String key = (String) keys.get(i);
            SubscribeInfo subscribeInfo = (SubscribeInfo) redis.get(key);
            result.add(subscribeInfo);
        }
        return result;
    }

    @Override
    public List<String> getAllSubscribePlatform() {
        String scanKey = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() +  "_Catalog_*";
        List<String> result = new ArrayList<>();
        List<Object> keys = redis.scan(scanKey);
        for (int i = 0; i < keys.size(); i++) {
            String key = (String) keys.get(i);
            String platformId = key.substring(scanKey.length() - 1);
            result.add(platformId);
        }
        return result;
    }
}

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

@@ -1,6 +1,9 @@
package com.genersoft.iot.vmp.storager.impl;

import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
@@ -27,15 +30,21 @@
import java.util.*;

/**    
 * @description:视频设备数据存储-jdbc实现
 * @author: swwheihei
 * @date:   2020年5月6日 下午2:31:42
 * 视频设备数据存储-jdbc实现
 * swwheihei
 * 2020年5月6日 下午2:31:42
 */
@SuppressWarnings("rawtypes")
@Component
public class VideoManagerStoragerImpl implements IVideoManagerStorager {

    private Logger logger = LoggerFactory.getLogger(VideoManagerStoragerImpl.class);

    @Autowired
    EventPublisher eventPublisher;

    @Autowired
    SipConfig sipConfig;

    @Autowired
    DataSourceTransactionManager dataSourceTransactionManager;
@@ -134,6 +143,7 @@
            return deviceMapper.add(device) > 0;
        }else {
            redisCatchStorage.updateDevice(device);

            return deviceMapper.update(device) > 0;
        }

@@ -408,6 +418,8 @@
        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;
    }

@@ -514,7 +526,7 @@
                if (parentPlatform.isShareAllLiveStream()) {
                    gbStreamService.addPlatformInfo(gbStreams, parentPlatform.getServerGBId(), parentPlatform.getCatalogId());
                }else {
                    gbStreamService.delPlatformInfo(gbStreams);
                    gbStreamService.delPlatformInfo(parentPlatform.getServerGBId(), gbStreams);
                }
            }
        }
@@ -590,6 +602,9 @@
        int result = 0;
        if (channelReducesToAdd.size() > 0) {
            result = platformChannelMapper.addChannels(platformId, channelReducesToAdd);
            // TODO 后续给平台增加控制开关以控制是否响应目录订阅
            List<DeviceChannel> deviceChannelList = getDeviceChannelListByChannelReduceList(channelReducesToAdd, catalogId);
            eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.ADD);
        }

        return result;
@@ -600,7 +615,13 @@
    public int delChannelForGB(String platformId, List<ChannelReduce> channelReduces) {

        int result = platformChannelMapper.delChannelForGB(platformId, channelReduces);

        List<DeviceChannel> deviceChannelList = new ArrayList<>();
        for (ChannelReduce channelReduce : channelReduces) {
            DeviceChannel deviceChannel = new DeviceChannel();
            deviceChannel.setChannelId(channelReduce.getChannelId());
            deviceChannelList.add(deviceChannel);
        }
        eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
        return result;
    }

@@ -739,7 +760,7 @@
     * @return
     */
    @Override
    public List<GbStream> queryStreamInParentPlatform(String platformId, String gbId) {
    public GbStream queryStreamInParentPlatform(String platformId, String gbId) {
        return gbStreamMapper.queryStreamInPlatform(platformId, gbId);
    }

@@ -771,7 +792,11 @@
        streamPushMapper.addAll(streamPushItems);
        // TODO 待优化
        for (int i = 0; i < streamPushItems.size(); i++) {
            gbStreamMapper.setStatus(streamPushItems.get(i).getApp(), streamPushItems.get(i).getStream(), true);
            int onlineResult = gbStreamMapper.setStatus(streamPushItems.get(i).getApp(), streamPushItems.get(i).getStream(), true);
            if (onlineResult > 0) {
                // 发送上线通知
                eventPublisher.catalogEventPublishForStream(null, streamPushItems.get(i), CatalogEvent.ON);
            }
        }
    }

@@ -780,6 +805,7 @@
        streamPushMapper.del(streamPushItem.getApp(), streamPushItem.getStream());
        streamPushMapper.add(streamPushItem);
        gbStreamMapper.setStatus(streamPushItem.getApp(), streamPushItem.getStream(), true);

        if(!StringUtils.isEmpty(streamPushItem.getGbId() )){
            // 查找开启了全部直播流共享的上级平台
            List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream();
@@ -858,7 +884,12 @@

    @Override
    public int addCatalog(PlatformCatalog platformCatalog) {
        return catalogMapper.add(platformCatalog);
        int result = catalogMapper.add(platformCatalog);
        if (result > 0) {
            DeviceChannel deviceChannel = getDeviceChannelByCatalog(platformCatalog);
            eventPublisher.catalogEventPublish(platformCatalog.getPlatformId(), deviceChannel, CatalogEvent.ADD);
        }
        return result;
    }

    @Override
@@ -873,23 +904,56 @@
            List<PlatformCatalog> platformCatalogList = catalogMapper.selectByParentId(platformCatalog.getPlatformId(), platformCatalog.getId());
            for (PlatformCatalog catalog : platformCatalogList) {
                if (catalog.getChildrenCount() == 0) {
                    catalogMapper.del(catalog.getId());
                    platformGbStreamMapper.delByCatalogId(catalog.getId());
                    platformChannelMapper.delByCatalogId(catalog.getId());
                    delCatalogExecute(catalog.getId(), catalog.getPlatformId());
                }else {
                    delCatalog(catalog.getId());
                }
            }
        }
        int delresult =  catalogMapper.del(id);
        int delStreamresult = platformGbStreamMapper.delByCatalogId(id);
        int delChanneresult = platformChannelMapper.delByCatalogId(id);
        return delresult + delChanneresult + delStreamresult;
        return delCatalogExecute(id, platformCatalog.getPlatformId());
    }
    private int delCatalogExecute(String id, String platformId) {
        int delresult =  catalogMapper.del(id);
        DeviceChannel deviceChannelForCatalog = new DeviceChannel();
        if (delresult > 0){
            deviceChannelForCatalog.setChannelId(id);
            eventPublisher.catalogEventPublish(platformId, deviceChannelForCatalog, CatalogEvent.DEL);
        }

        List<GbStream> gbStreams = platformGbStreamMapper.queryChannelInParentPlatformAndCatalog(platformId, id);
        if (gbStreams.size() > 0){
            List<DeviceChannel> deviceChannelList = new ArrayList<>();
            for (GbStream gbStream : gbStreams) {
                DeviceChannel deviceChannel = new DeviceChannel();
                deviceChannel.setChannelId(gbStream.getGbId());
                deviceChannelList.add(deviceChannel);
            }
            eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
        }
        int delStreamresult = platformGbStreamMapper.delByCatalogId(id);
        List<PlatformCatalog> platformCatalogs = platformChannelMapper.queryChannelInParentPlatformAndCatalog(platformId, id);
        if (platformCatalogs.size() > 0){
            List<DeviceChannel> deviceChannelList = new ArrayList<>();
            for (PlatformCatalog platformCatalog : platformCatalogs) {
                DeviceChannel deviceChannel = new DeviceChannel();
                deviceChannel.setChannelId(platformCatalog.getId());
                deviceChannelList.add(deviceChannel);
            }
            eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
        }
        int delChannelresult = platformChannelMapper.delByCatalogId(id);
        return delresult + delChannelresult + delStreamresult;
    }


    @Override
    public int updateCatalog(PlatformCatalog platformCatalog) {
        return catalogMapper.update(platformCatalog);
        int result = catalogMapper.update(platformCatalog);
        if (result > 0) {
            DeviceChannel deviceChannel = getDeviceChannelByCatalog(platformCatalog);
            eventPublisher.catalogEventPublish(platformCatalog.getPlatformId(), deviceChannel, CatalogEvent.UPDATE);
        }
        return result;
    }

    @Override
@@ -905,11 +969,17 @@
    @Override
    public int delRelation(PlatformCatalog platformCatalog) {
        if (platformCatalog.getType() == 1) {
            DeviceChannel deviceChannel = new DeviceChannel();
            deviceChannel.setChannelId(platformCatalog.getId());
            eventPublisher.catalogEventPublish(platformCatalog.getPlatformId(), deviceChannel, CatalogEvent.DEL);
            return platformChannelMapper.delByCatalogIdAndChannelIdAndPlatformId(platformCatalog);
        }else if (platformCatalog.getType() == 2) {
            List<GbStream> gbStreams = platformGbStreamMapper.queryChannelInParentPlatformAndCatalog(platformCatalog.getPlatformId(), platformCatalog.getParentId());
            for (GbStream gbStream : gbStreams) {
                if (gbStream.getGbId().equals(platformCatalog.getId())) {
                    DeviceChannel deviceChannel = new DeviceChannel();
                    deviceChannel.setChannelId(gbStream.getGbId());
                    eventPublisher.catalogEventPublish(platformCatalog.getPlatformId(), deviceChannel, CatalogEvent.DEL);
                    return platformGbStreamMapper.delByAppAndStream(gbStream.getApp(), gbStream.getStream());
                }
            }
@@ -921,4 +991,57 @@
    public int updateStreamGPS(List<GPSMsgInfo> gpsMsgInfos) {
        return gbStreamMapper.updateStreamGPS(gpsMsgInfos);
    }

    private List<DeviceChannel> getDeviceChannelListByChannelReduceList(List<ChannelReduce> channelReduces, String catalogId) {
        List<DeviceChannel> deviceChannelList = new ArrayList<>();
        if (channelReduces.size() > 0){
            for (ChannelReduce channelReduce : channelReduces) {
                DeviceChannel deviceChannel = queryChannel(channelReduce.getDeviceId(), channelReduce.getChannelId());
                deviceChannel.setParental(1);
                deviceChannel.setParentId(catalogId);
                deviceChannelList.add(deviceChannel);
            }
        }
        return deviceChannelList;
    }

    private DeviceChannel getDeviceChannelByCatalog(PlatformCatalog catalog) {
        ParentPlatform parentPlatByServerGBId = platformMapper.getParentPlatByServerGBId(catalog.getPlatformId());
        DeviceChannel deviceChannel = new DeviceChannel();
        deviceChannel.setChannelId(catalog.getId());
        deviceChannel.setName(catalog.getName());
        deviceChannel.setLongitude(0.0);
        deviceChannel.setLatitude(0.0);
        deviceChannel.setDeviceId(parentPlatByServerGBId.getDeviceGBId());
        deviceChannel.setManufacture("wvp-pro");
        deviceChannel.setStatus(1);
        deviceChannel.setParental(1);
        deviceChannel.setParentId(catalog.getParentId());
        deviceChannel.setRegisterWay(1);
        deviceChannel.setCivilCode(sipConfig.getDomain());
        deviceChannel.setModel("live");
        deviceChannel.setOwner("wvp-pro");
        deviceChannel.setSecrecy("0");
        return deviceChannel;
    }

    @Override
    public List<DeviceChannel> queryOnlineChannelsByDeviceId(String deviceId) {
        return deviceChannelMapper.queryOnlineChannelsByDeviceId(deviceId);
    }

    @Override
    public List<ParentPlatform> queryPlatFormListForGBWithGBId(String channelId, List<String> platforms) {
        return platformChannelMapper.queryPlatFormListForGBWithGBId(channelId, platforms);
    }

    @Override
    public List<ParentPlatform> queryPlatFormListForStreamWithGBId(String app, String stream, List<String> platforms) {
        return platformGbStreamMapper.queryPlatFormListForGBWithGBId(app, stream, platforms);
    }

    @Override
    public GbStream getGbStream(String app, String streamId) {
        return gbStreamMapper.selectOne(app, streamId);
    }
}

 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/gbStream/GbStreamController.java

@@ -62,7 +62,7 @@
    @DeleteMapping(value = "/del")
    @ResponseBody
    public Object del(@RequestBody GbStreamParam gbStreamParam){
        if (gbStreamService.delPlatformInfo(gbStreamParam.getGbStreams())) {
        if (gbStreamService.delPlatformInfo(gbStreamParam.getPlatformId(), gbStreamParam.getGbStreams())) {
            return "success";
        }else {
            return "fail";

 web_src/package-lock.json

@@ -5094,7 +5094,8 @@
    },
    "js-yaml": {
      "version": "3.7.0",
      "resolved": "",
      "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-3.7.0.tgz",
      "integrity": "sha1-XJZ93YN6m/3KXy3oQlOr6KHAO4A=",
      "dev": true,
      "requires": {
        "argparse": "^1.0.7",

 web_src/src/components/dialog/chooseChannelForStream.vue

@@ -147,6 +147,7 @@
                    method:"delete",
                    url:"/api/gbStream/del",
                    data:{
                        platformId: that.platformId,
                        gbStreams:  delData,
                    }
                }).then((res)=>{