Merge branch 'wvp-28181-2.0' into main-dev

648540858

2023-10-11 7de73ebd2bc07a51f0f9db031d6f616bdcfe549c

Merge branch 'wvp-28181-2.0' into main-dev

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
#	src/main/java/com/genersoft/iot/vmp/service/IPlatformService.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java

 pom.xml

@@ -207,6 +207,12 @@
            <version>2.1.3</version>
        </dependency>

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>20.0</version>
        </dependency>

        <!-- json解析库fastjson2 -->
        <dependency>
            <groupId>com.alibaba.fastjson2</groupId>

 sql/2.6.9¸üÐÂ.sql

@@ -1,2 +1,8 @@
alter table wvp_device_channel
    change stream_id stream_id varying(255)
    change stream_id stream_id varying(255)

alter table wvp_platform
    add auto_push_channel bool default false

alter table wvp_stream_proxy
    add stream_key varying(255)

 sql/³õʼ»¯.sql

@@ -194,6 +194,7 @@
                              create_time character varying(50),
                              update_time character varying(50),
                              as_message_channel bool default false,
                              auto_push_channel bool default false,
                              constraint uk_platform_unique_server_gb_id unique (server_gb_id)
);

@@ -243,6 +244,7 @@
                                  create_time character varying(50),
                                  name character varying(255),
                                  update_time character varying(50),
                                  stream_key character varying(255),
                                  enable_disable_none_reader bool default false,
                                  constraint uk_stream_proxy_app_stream unique (app, stream)
);

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

@@ -186,6 +186,9 @@
    @Schema(description = "是否作为消息通道")
    private boolean asMessageChannel;

    @Schema(description = "是否作为消息通道")
    private boolean autoPushChannel;

    public Integer getId() {
        return id;
    }
@@ -425,4 +428,12 @@
    public void setAsMessageChannel(boolean asMessageChannel) {
        this.asMessageChannel = asMessageChannel;
    }

    public boolean isAutoPushChannel() {
        return autoPushChannel;
    }

    public void setAutoPushChannel(boolean autoPushChannel) {
        this.autoPushChannel = autoPushChannel;
    }
}

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

@@ -32,11 +32,13 @@

    public void putCatalogSubscribe(String platformId, SubscribeInfo subscribeInfo) {
        catalogMap.put(platformId, subscribeInfo);
        // æ·»åŠ è®¢é˜…到期
        String taskOverdueKey = taskOverduePrefix +  "catalog_" + platformId;
        // æ·»åŠ ä»»åŠ¡å¤„理订阅过期
        dynamicTask.startDelay(taskOverdueKey, () -> removeCatalogSubscribe(subscribeInfo.getId()),
                subscribeInfo.getExpires() * 1000);
        if (subscribeInfo.getExpires() > 0) {
            // æ·»åŠ è®¢é˜…到期
            String taskOverdueKey = taskOverduePrefix +  "catalog_" + platformId;
            // æ·»åŠ ä»»åŠ¡å¤„理订阅过期
            dynamicTask.startDelay(taskOverdueKey, () -> removeCatalogSubscribe(subscribeInfo.getId()),
                    subscribeInfo.getExpires() * 1000);
        }
    }

    public SubscribeInfo getCatalogSubscribe(String platformId) {
@@ -63,11 +65,13 @@
        dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(platformId),
                subscribeInfo.getGpsInterval() * 1000);
        String taskOverdueKey = taskOverduePrefix +  "MobilePosition_" + platformId;
        // æ·»åŠ ä»»åŠ¡å¤„理订阅过期
        dynamicTask.startDelay(taskOverdueKey, () -> {
                    removeMobilePositionSubscribe(subscribeInfo.getId());
                },
                subscribeInfo.getExpires() * 1000);
        if (subscribeInfo.getExpires() > 0) {
            // æ·»åŠ ä»»åŠ¡å¤„理订阅过期
            dynamicTask.startDelay(taskOverdueKey, () -> {
                        removeMobilePositionSubscribe(subscribeInfo.getId());
                    },
                    subscribeInfo.getExpires() * 1000);
        }
    }

    public SubscribeInfo getMobilePositionSubscribe(String platformId) {

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

@@ -18,6 +18,9 @@

    }

    public SubscribeInfo() {
    }

    private String id;

    private SIPRequest request;
@@ -32,6 +35,21 @@
     */
    private String sn;
    private int gpsInterval;

    /**
     * æ¨¡æ‹Ÿçš„FromTag
     */
    private String simulatedFromTag;

    /**
     * æ¨¡æ‹Ÿçš„ToTag
     */
    private String simulatedToTag;

    /**
     * æ¨¡æ‹Ÿçš„CallID
     */
    private String simulatedCallId;

    public String getId() {
        return id;
@@ -96,4 +114,28 @@
    public void setGpsInterval(int gpsInterval) {
        this.gpsInterval = gpsInterval;
    }

    public String getSimulatedFromTag() {
        return simulatedFromTag;
    }

    public void setSimulatedFromTag(String simulatedFromTag) {
        this.simulatedFromTag = simulatedFromTag;
    }

    public String getSimulatedCallId() {
        return simulatedCallId;
    }

    public void setSimulatedCallId(String simulatedCallId) {
        this.simulatedCallId = simulatedCallId;
    }

    public String getSimulatedToTag() {
        return simulatedToTag;
    }

    public void setSimulatedToTag(String simulatedToTag) {
        this.simulatedToTag = simulatedToTag;
    }
}

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

@@ -93,7 +93,10 @@
                    }
                    if (event.getGbStreams() != null && event.getGbStreams().size() > 0){
                        for (GbStream gbStream : event.getGbStreams()) {
                            if (gbStream.getStreamType().equals("push") && !userSetting.isUsePushingAsStatus()) {
                            if (gbStream != null
                                    && gbStream.getStreamType() != null
                                    && gbStream.getStreamType().equals("push")
                                    && !userSetting.isUsePushingAsStatus()) {
                                continue;
                            }
                            DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform);

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

@@ -228,11 +228,11 @@
        SipURI fromSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(parentPlatform.getDeviceGBId(),
                parentPlatform.getDeviceIp() + ":" + parentPlatform.getDevicePort());
        Address fromAddress = SipFactory.getInstance().createAddressFactory().createAddress(fromSipURI);
        FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, subscribeInfo.getResponse().getToTag());
        FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, subscribeInfo.getResponse() != null ? subscribeInfo.getResponse().getToTag(): subscribeInfo.getSimulatedToTag());
        // to
        SipURI toSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerGBDomain());
        Address toAddress = SipFactory.getInstance().createAddressFactory().createAddress(toSipURI);
        ToHeader toHeader = SipFactory.getInstance().createHeaderFactory().createToHeader(toAddress, subscribeInfo.getRequest().getFromTag());
        ToHeader toHeader = SipFactory.getInstance().createHeaderFactory().createToHeader(toAddress, subscribeInfo.getRequest() != null ?subscribeInfo.getRequest().getFromTag(): subscribeInfo.getSimulatedFromTag());

        // Forwards
        MaxForwardsHeader maxForwards = SipFactory.getInstance().createHeaderFactory().createMaxForwardsHeader(70);
@@ -242,7 +242,7 @@
        // è®¾ç½®ç¼–码， é˜²æ­¢ä¸­æ–‡ä¹±ç 
        messageFactory.setDefaultContentEncodingCharset("gb2312");

        CallIdHeader callIdHeader = SipFactory.getInstance().createHeaderFactory().createCallIdHeader(subscribeInfo.getRequest().getCallIdHeader().getCallId());
        CallIdHeader callIdHeader = SipFactory.getInstance().createHeaderFactory().createCallIdHeader(subscribeInfo.getRequest() != null ? subscribeInfo.getRequest().getCallIdHeader().getCallId(): subscribeInfo.getSimulatedCallId());

        request = (SIPRequest) messageFactory.createRequest(requestURI, Request.NOTIFY, callIdHeader, cSeqHeader, fromHeader,
                toHeader, viaHeaders, maxForwards);

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

@@ -169,13 +169,13 @@

        CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());

            Request request = headerProviderPlatformProvider.createMessageRequest(
                    parentPlatform,
                    keepaliveXml.toString(),
                    SipUtils.getNewFromTag(),
                    SipUtils.getNewViaTag(),
                    callIdHeader);
            sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, errorEvent, okEvent);
        Request request = headerProviderPlatformProvider.createMessageRequest(
                parentPlatform,
                keepaliveXml.toString(),
                SipUtils.getNewFromTag(),
                SipUtils.getNewViaTag(),
                callIdHeader);
        sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, errorEvent, okEvent);
        return callIdHeader.getCallId();
    }


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

@@ -3,6 +3,7 @@
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.google.common.primitives.Bytes;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
import org.apache.commons.lang3.ArrayUtils;
@@ -203,15 +204,14 @@
                result.add(rawContent[i]);
            }
        }
        Byte[] bytes = new Byte[0];
        byte[] bytesResult = ArrayUtils.toPrimitive(result.toArray(bytes));
        byte[] bytesResult = Bytes.toArray(result);

        Document xml;
        try {
            xml = reader.read(new ByteArrayInputStream(bytesResult));
        }catch (DocumentException e) {
            logger.warn("[xml解析异常]： æ„¿æ–‡å¦‚下： \r\n{}", new String(bytesResult));
            logger.warn("[xml解析异常]： æ„¿æ–‡å¦‚下： å°è¯•å…¼å®¹æ€§å¤„理");
            logger.warn("[xml解析异常]： åŽŸæ–‡å¦‚下： \r\n{}", new String(bytesResult));
            logger.warn("[xml解析异常]： åŽŸæ–‡å¦‚下： å°è¯•å…¼å®¹æ€§å¤„理");
            String[] xmlLineArray = new String(bytesResult).split("\\r?\\n");

            // å…¼å®¹æµ·åº·çš„address字段带有<破换xml结构导致无法解析xml的问题

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

@@ -13,6 +13,7 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
@@ -52,6 +53,10 @@

    @Autowired
    private SIPSender sipSender;


    @Autowired
    private IPlatformService platformService;

    @Override
    public void afterPropertiesSet() throws Exception {
@@ -194,5 +199,8 @@
        } catch (SipException | InvalidArgumentException | ParseException e) {
            logger.error("未处理的异常 ", e);
        }
        if (subscribeHolder.getCatalogSubscribe(platformId) == null && platform.isAutoPushChannel()) {
            platformService.addSimulatedSubscribeInfo(platform);
        }
    }
}

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

@@ -78,7 +78,6 @@
        }
        taskExecutor.execute(()->{
            try {

                String sn = getText(rootElement, "SN");
                String channelId = getText(rootElement, "DeviceID");
                RecordInfo recordInfo = new RecordInfo();

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

@@ -206,6 +206,13 @@
        }

        // æŽ¨æµé‰´æƒçš„处理

        if (!"rtp".equals(param.getApp())) {

            StreamProxyItem stream = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());

            if (stream != null) {

                HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();

                result.setEnable_audio(stream.isEnableAudio());

                result.setEnable_mp4(stream.isEnableMp4());

                return result;

            }

            if (userSetting.getPushAuthority()) {

                // æŽ¨æµé‰´æƒ

                if (param.getParams() == null) {


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

@@ -32,13 +32,20 @@
    }

    private OkHttpClient getClient(){
        return getClient(null);
    }

    private OkHttpClient getClient(Integer readTimeOut){
        if (client == null) {
            if (readTimeOut == null) {
                readTimeOut = 10;
            }
            OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
            //todo æš‚时写死超时时间 å‡ä¸º5s
            // è®¾ç½®è¿žæŽ¥è¶…时时间
            httpClientBuilder.connectTimeout(5,TimeUnit.SECONDS);
            httpClientBuilder.connectTimeout(8,TimeUnit.SECONDS);
            // è®¾ç½®è¯»å–超时时间
            httpClientBuilder.readTimeout(10,TimeUnit.SECONDS);
            httpClientBuilder.readTimeout(readTimeOut,TimeUnit.SECONDS);
            // è®¾ç½®è¿žæŽ¥æ± 
            httpClientBuilder.connectionPool(new ConnectionPool(16, 5, TimeUnit.MINUTES));
            if (logger.isDebugEnabled()) {
@@ -55,9 +62,13 @@

    }


    public JSONObject sendPost(MediaServerItem mediaServerItem, String api, Map<String, Object> param, RequestCallback callback) {
        OkHttpClient client = getClient();
        return sendPost(mediaServerItem, api, param, callback, null);
    }


    public JSONObject sendPost(MediaServerItem mediaServerItem, String api, Map<String, Object> param, RequestCallback callback, Integer readTimeOut) {
        OkHttpClient client = getClient(readTimeOut);

        if (mediaServerItem == null) {
            return null;
@@ -264,6 +275,12 @@
        return sendPost(mediaServerItem, "delFFmpegSource",param, null);
    }

    public JSONObject delStreamProxy(MediaServerItem mediaServerItem, String key){
        Map<String, Object> param = new HashMap<>();
        param.put("key", key);
        return sendPost(mediaServerItem, "delStreamProxy",param, null);
    }

    public JSONObject getMediaServerConfig(MediaServerItem mediaServerItem){
        return sendPost(mediaServerItem, "getServerConfig",null, null);
    }
@@ -317,7 +334,7 @@
        param.put("enable_mp4", enable_mp4?1:0);
        param.put("enable_audio", enable_audio?1:0);
        param.put("rtp_type", rtp_type);
        return sendPost(mediaServerItem, "addStreamProxy",param, null);
        return sendPost(mediaServerItem, "addStreamProxy",param, null, 20);
    }

    public JSONObject closeStreams(MediaServerItem mediaServerItem, String app, String stream) {

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java

@@ -41,6 +41,9 @@
    @Schema(description = "是否 æ— äººè§‚看时自动停用")
    private boolean enableDisableNoneReader;

    @Schema(description = "拉流代理时zlm返回的key，用于停止拉流代理")
    private String streamKey;

    public String getType() {
        return type;
    }
@@ -167,5 +170,11 @@
        this.enableAudio = enable_audio;
    }

    public String getStreamKey() {
        return streamKey;
    }

    public void setStreamKey(String streamKey) {
        this.streamKey = streamKey;
    }
}

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

@@ -69,4 +69,6 @@
     * @param catalogId
     */
    void delAllPlatformInfo(String platformId, String catalogId);

    List<GbStream> getGbChannelWithGbid(String gbId);
}

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

@@ -80,4 +80,6 @@
     * è¯­éŸ³å–Šè¯å›žå¤BYE
     */
    void stopBroadcast(ParentPlatform platform, DeviceChannel channel, String stream,boolean sendBye, MediaServerItem mediaServerItem);

    void addSimulatedSubscribeInfo(ParentPlatform parentPlatform);
}

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

@@ -18,6 +18,7 @@
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;

import java.util.ArrayList;
@@ -263,4 +264,9 @@
            eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
        }
    }

    @Override
    public List<GbStream> getGbChannelWithGbid(String gbId) {
        return gbStreamMapper.selectByGBId(gbId);
    }
}

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

@@ -13,6 +13,7 @@
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
@@ -28,6 +29,7 @@
import com.genersoft.iot.vmp.utils.DateUtil;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import gov.nist.javax.sip.message.SIPRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,12 +37,19 @@

import javax.sip.InvalidArgumentException;
import javax.sip.ResponseEvent;
import javax.sip.PeerUnavailableException;
import javax.sip.SipException;
import javax.sip.SipFactory;
import javax.sip.address.Address;
import javax.sip.address.SipURI;
import javax.sip.header.*;
import javax.sip.message.Request;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.*;

/**
 * @author lin
@@ -199,6 +208,7 @@
            }
        }


        return false;
    }

@@ -243,18 +253,19 @@
                        try {
                            commanderForPlatform.keepalive(parentPlatform, eventResult -> {
                                // å¿ƒè·³å¤±è´¥
                                if (eventResult.type == SipSubscribe.EventResultType.timeout) {
                                    // å¿ƒè·³è¶…æ—¶
                                    ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
                                    // æ­¤æ—¶æ˜¯ç¬¬ä¸‰æ¬¡å¿ƒè·³è¶…时， å¹³å°ç¦»çº¿
                                    if (platformCatch.getKeepAliveReply()  == 2) {
                                        // è®¾ç½®å¹³å°ç¦»çº¿ï¼Œå¹¶é‡æ–°æ³¨å†Œ
                                        logger.info("[国标级联] ä¸‰æ¬¡å¿ƒè·³è¶…æ—¶, å¹³å°{}({})离线", parentPlatform.getName(), parentPlatform.getServerGBId());
                                        offline(parentPlatform, false);
                                    }

                                }else {
                                if (eventResult.type != SipSubscribe.EventResultType.timeout) {
                                    logger.warn("[国标级联]发送心跳收到错误，code： {}, msg: {}", eventResult.statusCode, eventResult.msg);
                                }
                                // å¿ƒè·³å¤±è´¥
                                ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
                                // æ­¤æ—¶æ˜¯ç¬¬ä¸‰æ¬¡å¿ƒè·³è¶…时， å¹³å°ç¦»çº¿
                                if (platformCatch.getKeepAliveReply()  == 2) {
                                    // è®¾ç½®å¹³å°ç¦»çº¿ï¼Œå¹¶é‡æ–°æ³¨å†Œ
                                    logger.info("[国标级联] ä¸‰æ¬¡å¿ƒè·³å¤±è´¥, å¹³å°{}({})离线", parentPlatform.getName(), parentPlatform.getServerGBId());
                                    offline(parentPlatform, false);
                                }else {
                                    platformCatch.setKeepAliveReply(platformCatch.getKeepAliveReply() + 1);
                                    redisCatchStorage.updatePlatformCatchInfo(platformCatch);
                                }

                            }, eventResult -> {
@@ -273,6 +284,31 @@
                    },
                    (parentPlatform.getKeepTimeout())*1000);
        }
        if (parentPlatform.isAutoPushChannel()) {
            if (subscribeHolder.getCatalogSubscribe(parentPlatform.getServerGBId()) == null) {
                addSimulatedSubscribeInfo(parentPlatform);
            }
        }else {
            SubscribeInfo catalogSubscribe = subscribeHolder.getCatalogSubscribe(parentPlatform.getServerGBId());
            if (catalogSubscribe != null && catalogSubscribe.getExpires() == -1) {
                subscribeHolder.removeCatalogSubscribe(parentPlatform.getServerGBId());
            }
        }
    }

    @Override
    public void addSimulatedSubscribeInfo(ParentPlatform parentPlatform) {
        // è‡ªåŠ¨æ·»åŠ ä¸€æ¡æ¨¡æ‹Ÿçš„订阅信息
        SubscribeInfo subscribeInfo = new SubscribeInfo();
        subscribeInfo.setId(parentPlatform.getServerGBId());
        subscribeInfo.setExpires(-1);
        subscribeInfo.setEventType("Catalog");
        int random = (int) Math.floor(Math.random() * 10000);
        subscribeInfo.setEventId(random + "");
        subscribeInfo.setSimulatedCallId(UUID.randomUUID().toString().replace("-", "") + "@" + parentPlatform.getServerIP());
        subscribeInfo.setSimulatedFromTag(UUID.randomUUID().toString().replace("-", ""));
        subscribeInfo.setSimulatedToTag(UUID.randomUUID().toString().replace("-", ""));
        subscribeHolder.putCatalogSubscribe(parentPlatform.getServerGBId(), subscribeInfo);
    }

    private void registerTask(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo){

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

@@ -10,7 +10,10 @@
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.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
@@ -58,6 +61,9 @@

    @Autowired
    private ZLMRESTfulUtils zlmresTfulUtils;

    @Autowired
    private ZLMServerFactory zlmServerFactory;

    @Autowired
    private StreamProxyMapper streamProxyMapper;
@@ -137,7 +143,7 @@
            dstUrl = String.format("%s://%s:%s/%s/%s", schemaForUri, "127.0.0.1", port, param.getApp(),
                    param.getStream());
        }else {
            dstUrl = String.format("rtmp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtmpPort(), param.getApp(),
            dstUrl = String.format("rtsp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtspPort(), param.getApp(),
                    param.getStream());
        }
        param.setDstUrl(dstUrl);
@@ -154,15 +160,14 @@
            callback.run(ErrorCode.ERROR100.getCode(), "保存失败", null);
            return;
        }

        HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaInfo.getId());
        hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
                    mediaInfo, param.getApp(), param.getStream(), null, null);
            callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
        });
        if (param.isEnable()) {
            String talkKey = UUID.randomUUID().toString();
            dynamicTask.startCron(talkKey, ()->{
                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaInfo.getId(), false);
                if (streamInfo != null) {
                    callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
                }
            }, 1000);
            String delayTalkKey = UUID.randomUUID().toString();
            dynamicTask.startDelay(delayTalkKey, ()->{
                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaInfo.getId(), false);
@@ -172,9 +177,10 @@
                    dynamicTask.stop(talkKey);
                    callback.run(ErrorCode.ERROR100.getCode(), "超时", null);
                }
            }, 5000);
            }, 7000);
            JSONObject jsonObject = addStreamProxyToZlm(param);
            if (jsonObject != null && jsonObject.getInteger("code") == 0) {
                hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
                dynamicTask.stop(talkKey);
                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
                        mediaInfo, param.getApp(), param.getStream(), null, null);
@@ -304,13 +310,32 @@
        if (mediaServerItem == null) {
            return null;
        }
        if ("default".equals(param.getType())){
            result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl().trim(),
                    param.isEnableAudio(), param.isEnableMp4(), param.getRtpType());
        }else if ("ffmpeg".equals(param.getType())) {
        if (zlmServerFactory.isStreamReady(mediaServerItem, param.getApp(), param.getStream())) {
            zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream());
        }
        if ("ffmpeg".equalsIgnoreCase(param.getType())){
            result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrcUrl().trim(), param.getDstUrl(),
                    param.getTimeoutMs() + "", param.isEnableAudio(), param.isEnableMp4(),
                    param.getFfmpegCmdKey());
        }else {
            result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl().trim(),
                    param.isEnableAudio(), param.isEnableMp4(), param.getRtpType());
        }
        System.out.println("addStreamProxyToZlm====");
        System.out.println(result);
        if (result != null && result.getInteger("code") == 0) {
            JSONObject data = result.getJSONObject("data");
            if (data == null) {
                logger.warn("[获取拉流代理的结果数据Data] å¤±è´¥ï¼š {}", result );
                return result;
            }
            String key = data.getString("key");
            if (key == null) {
                logger.warn("[获取拉流代理的结果数据Data中的KEY] å¤±è´¥ï¼š {}", result );
                return result;
            }
            param.setStreamKey(key);
            streamProxyMapper.update(param);
        }
        return result;
    }
@@ -321,7 +346,12 @@
            return null;
        }
        MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
        JSONObject result = zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream());
        JSONObject result = null;
        if ("ffmpeg".equalsIgnoreCase(param.getType())){
            result = zlmresTfulUtils.delFFmpegSource(mediaServerItem, param.getStreamKey());
        }else {
            result = zlmresTfulUtils.delStreamProxy(mediaServerItem, param.getStreamKey());
        }
        return result;
    }

@@ -335,18 +365,19 @@
        StreamProxyItem streamProxyItem = videoManagerStorager.queryStreamProxy(app, stream);
        if (streamProxyItem != null) {
            gbStreamService.sendCatalogMsg(streamProxyItem, CatalogEvent.DEL);

            // å¦‚果关联了国标那么移除关联
            platformGbStreamMapper.delByAppAndStream(app, stream);
            gbStreamMapper.del(app, stream);
            videoManagerStorager.deleteStreamProxy(app, stream);
            redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream);
            JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
            if (jsonObject != null && jsonObject.getInteger("code") == 0) {
                // å¦‚果关联了国标那么移除关联
                gbStreamMapper.del(app, stream);
                platformGbStreamMapper.delByAppAndStream(app, stream);
                // TODO å¦‚果关联的推流， é‚£ä¹ˆçŠ¶æ€è®¾ç½®ä¸ºç¦»çº¿
                logger.info("[移除代理]： ä»£ç†ï¼š {}/{}, ä»Žzlm移除成功", app, stream);
            }else {
                logger.info("[移除代理]： ä»£ç†ï¼š {}/{}, ä»Žzlm移除失败", app, stream);
            }
            redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream);
        }


    }

    @Override

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

@@ -440,7 +440,7 @@

                }
            }
            if (streamPushItemListFroPlatform.size() > 0) {
            if (!streamPushItemListFroPlatform.isEmpty()) {
                platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
                // å‘送通知
                for (String platformId : platformForEvent.keySet()) {

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

@@ -167,8 +167,8 @@
            " <if test='query != null'> AND (dc.channel_id LIKE concat('%',#{query},'%') OR dc.name LIKE concat('%',#{query},'%') OR dc.name LIKE concat('%',#{query},'%'))</if> " +
            " <if test='online == true' > AND dc.status=true</if> " +
            " <if test='online == false' > AND dc.status=false</if> " +
            " <if test='hasSubChannel!= null and has_sub_channel == true' >  AND dc.sub_count > 0</if> " +
            " <if test='hasSubChannel!= null and has_sub_channel == false' >  AND dc.sub_count = 0</if> " +
            " <if test='hasSubChannel!= null and hasSubChannel == true' >  AND dc.sub_count > 0</if> " +
            " <if test='hasSubChannel!= null and hasSubChannel == false' >  AND dc.sub_count = 0</if> " +
            " <if test='catalogId == null ' >  AND dc.id not in (select device_channel_id from wvp_platform_gb_channel where platform_id=#{platformId} ) </if> " +
            " <if test='catalogId != null ' >  AND pgc.platform_id = #{platformId} and pgc.catalog_id=#{catalogId} </if> " +
            " ORDER BY dc.device_id, dc.channel_id ASC" +

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

@@ -16,10 +16,10 @@
public interface ParentPlatformMapper {

    @Insert("INSERT INTO wvp_platform (enable, name, server_gb_id, server_gb_domain, server_ip, server_port,device_gb_id,device_ip,"+
            "device_port,username,password,expires,keep_timeout,transport,character_set,ptz,rtcp,as_message_channel,"+
            "device_port,username,password,expires,keep_timeout,transport,character_set,ptz,rtcp,as_message_channel,auto_push_channel,"+
            "status,start_offline_push,catalog_id,administrative_division,catalog_group,create_time,update_time) " +
            "            VALUES (#{enable}, #{name}, #{serverGBId}, #{serverGBDomain}, #{serverIP}, #{serverPort}, #{deviceGBId}, #{deviceIp}, " +
            "            #{devicePort}, #{username}, #{password}, #{expires}, #{keepTimeout}, #{transport}, #{characterSet}, #{ptz}, #{rtcp}, #{asMessageChannel}, " +
            "            #{devicePort}, #{username}, #{password}, #{expires}, #{keepTimeout}, #{transport}, #{characterSet}, #{ptz}, #{rtcp}, #{asMessageChannel}, #{autoPushChannel}, " +
            "            #{status},  #{startOfflinePush}, #{catalogId}, #{administrativeDivision}, #{catalogGroup}, #{createTime}, #{updateTime})")
    int addParentPlatform(ParentPlatform parentPlatform);

@@ -42,6 +42,7 @@
            "ptz=#{ptz}, " +
            "rtcp=#{rtcp}, " +
            "as_message_channel=#{asMessageChannel}, " +
            "auto_push_channel=#{autoPushChannel}, " +
            "status=#{status}, " +
            "start_offline_push=#{startOfflinePush}, " +
            "catalog_group=#{catalogGroup}, " +

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

@@ -12,9 +12,9 @@
public interface StreamProxyMapper {

    @Insert("INSERT INTO wvp_stream_proxy (type, name, app, stream,media_server_id, url, src_url, dst_url, " +
            "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_audio, enable_mp4, enable, status, enable_remove_none_reader, enable_disable_none_reader, create_time) VALUES" +
            "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_audio, enable_mp4, enable, status, stream_key, enable_remove_none_reader, enable_disable_none_reader, create_time) VALUES" +
            "(#{type}, #{name}, #{app}, #{stream}, #{mediaServerId}, #{url}, #{srcUrl}, #{dstUrl}, " +
            "#{timeoutMs}, #{ffmpegCmdKey}, #{rtpType}, #{enableAudio}, #{enableMp4}, #{enable}, #{status}, " +
            "#{timeoutMs}, #{ffmpegCmdKey}, #{rtpType}, #{enableAudio}, #{enableMp4}, #{enable}, #{status}, #{streamKey}, " +
            "#{enableRemoveNoneReader}, #{enableDisableNoneReader}, #{createTime} )")
    int add(StreamProxyItem streamProxyDto);

@@ -33,6 +33,7 @@
            "enable_audio=#{enableAudio}, " +
            "enable=#{enable}, " +
            "status=#{status}, " +
            "stream_key=#{streamKey}, " +
            "enable_remove_none_reader=#{enableRemoveNoneReader}, " +
            "enable_disable_none_reader=#{enableDisableNoneReader}, " +
            "enable_mp4=#{enableMp4} " +
@@ -45,7 +46,7 @@
    @Select("SELECT st.*, pgs.gb_id, pgs.name, pgs.longitude, pgs.latitude FROM wvp_stream_proxy st LEFT join wvp_gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream order by st.create_time desc")
    List<StreamProxyItem> selectAll();

    @Select("SELECT st.*, pgs.gb_id, pgs.name, pgs.longitude, pgs.latitude FROM wvp_stream_proxy st LEFT join wvp_gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream WHERE st.enable=#{enable} order by st.create_time desc")
    @Select("SELECT st.*, pgs.gb_id, pgs.name, pgs.longitude, pgs.latitude, 'proxy' as streamType FROM wvp_stream_proxy st LEFT join wvp_gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream WHERE st.enable=#{enable} order by st.create_time desc")
    List<StreamProxyItem> selectForEnable(boolean enable);

    @Select("SELECT st.*, pgs.gb_id, pgs.name, pgs.longitude, pgs.latitude FROM wvp_stream_proxy st LEFT join wvp_gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream WHERE st.app=#{app} AND st.stream=#{stream} order by st.create_time desc")

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

@@ -36,6 +36,7 @@
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;

import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
@@ -472,7 +473,10 @@
        try {
            final InputStream in = Files.newInputStream(new File("snap" + File.separator + deviceId + "_" + channelId + (mark == null? ".jpg": ("_" + mark + ".jpg"))).toPath());
            resp.setContentType(MediaType.IMAGE_PNG_VALUE);
            ServletOutputStream outputStream = resp.getOutputStream();
            IOUtils.copy(in, resp.getOutputStream());
            in.close();
            outputStream.close();
        } catch (IOException e) {
            resp.setStatus(HttpServletResponse.SC_NOT_FOUND);
        }

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

@@ -1,8 +1,12 @@
package com.genersoft.iot.vmp.vmanager.gb28181.gbStream;

import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.gb28181.gbStream.bean.GbStreamParam;
import com.github.pagehelper.PageInfo;
import io.swagger.v3.oas.annotations.Operation;
@@ -14,6 +18,7 @@
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*;

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

@Tag(name  = "视频流关联到级联平台")
@@ -28,7 +33,7 @@
    private IGbStreamService gbStreamService;

    @Autowired
    private IVideoManagerStorage storager;
    private IPlatformService platformService;


    /**
@@ -107,4 +112,20 @@
            gbStreamService.addPlatformInfo(gbStreamParam.getGbStreams(), gbStreamParam.getPlatformId(), gbStreamParam.getCatalogId());
        }
    }

    /**
     * ä¿å­˜å›½æ ‡å…³è”
     * @param gbId
     * @return
     */
    @Operation(summary = "保存国标关联")
    @GetMapping(value = "/addWithGbid")
    @ResponseBody
    public void add(String gbId, String platformGbId, @RequestParam(required = false) String catalogGbId){
        List<GbStream> gbStreams = gbStreamService.getGbChannelWithGbid(gbId);
        if (gbStreams.isEmpty()) {
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "gbId的信息未找到");
        }
        gbStreamService.addPlatformInfo(gbStreams, platformGbId, catalogGbId);
    }
}

 src/main/java/com/genersoft/iot/vmp/vmanager/streamProxy/StreamProxyController.java

@@ -67,6 +67,16 @@
        return streamProxyService.getAll(page, count);
    }

    @Operation(summary = "查询流代理")
    @Parameter(name = "app", description = "应用名")
    @Parameter(name = "stream", description = "流Id")
    @GetMapping(value = "/one")
    @ResponseBody
    public StreamProxyItem one(String app, String stream){

        return streamProxyService.getStreamProxyByAppAndStream(app, stream);
    }

    @Operation(summary = "保存代理", parameters = {
            @Parameter(name = "param", description = "代理参数", required = true),
    })
@@ -80,9 +90,16 @@
        if (ObjectUtils.isEmpty(param.getType())) {
            param.setType("default");
        }
        if (ObjectUtils.isEmpty(param.getRtpType())) {
            param.setRtpType("1");
        }
        if (ObjectUtils.isEmpty(param.getGbId())) {
            param.setGbId(null);
        }
        StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());
        if (streamProxyItem  != null) {
            streamProxyService.del(param.getApp(), param.getStream());
        }

        RequestMessage requestMessage = new RequestMessage();
        String key = DeferredResultHolder.CALLBACK_CMD_PROXY + param.getApp() + param.getStream();

 web_src/src/components/dialog/platformEdit.vue

@@ -91,9 +91,10 @@
              <el-form-item label="其他选项">
                <el-checkbox label="启用" v-model="platform.enable" @change="checkExpires"></el-checkbox>
<!--                <el-checkbox label="云台控制" v-model="platform.ptz"></el-checkbox>-->
                <el-checkbox label="拉起离线推流" v-model="platform.startOfflinePush"></el-checkbox>
                <el-checkbox label="拉起推流" v-model="platform.startOfflinePush"></el-checkbox>
                <el-checkbox label="RTCP保活" v-model="platform.rtcp" @change="rtcpCheckBoxChange"></el-checkbox>
                <el-checkbox label="作为消息通道" v-model="platform.asMessageChannel" ></el-checkbox>
                <el-checkbox label="消息通道" v-model="platform.asMessageChannel" ></el-checkbox>
                <el-checkbox label="推送通道" v-model="platform.autoPushChannel" ></el-checkbox>
              </el-form-item>
              <el-form-item>
                <el-button type="primary" @click="onSubmit">{{
@@ -141,6 +142,7 @@
        ptz: true,
        rtcp: false,
        asMessageChannel: false,
        autoPushChannel: false,
        name: null,
        serverGBId: null,
        serverGBDomain: null,
@@ -208,6 +210,7 @@
        this.platform.ptz = platform.ptz;
        this.platform.rtcp = platform.rtcp;
        this.platform.asMessageChannel = platform.asMessageChannel;
        this.platform.autoPushChannel = platform.autoPushChannel;
        this.platform.name = platform.name;
        this.platform.serverGBId = platform.serverGBId;
        this.platform.serverGBDomain = platform.serverGBDomain;
@@ -284,6 +287,7 @@
        ptz: true,
        rtcp: false,
        asMessageChannel: false,
        autoPushChannel: false,
        name: null,
        serverGBId: null,
        administrativeDivision: null,