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

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
648540858
2 parents e45b4436 205f1f1f
Showing 28 changed files with 314 additions and 75 deletions
pom.xml
sql/2.6.9更新.sql
sql/初始化.sql
src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java
src/main/java/com/genersoft/iot/vmp/service/IGbStreamService.java
src/main/java/com/genersoft/iot/vmp/service/IPlatformService.java
src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
@@ -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>
 alter table wvp_device_channel
-    change stream_id stream_id varying(255)
 \ No newline at end of file
+    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)
@@ -194,6 +194,7 @@ create table wvp_platform (
                               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 table wvp_stream_proxy (
                                   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)
 );
@@ -186,6 +186,9 @@ public class ParentPlatform {
     @Schema(description = "是否作为消息通道")
     private boolean asMessageChannel;
+    @Schema(description = "是否作为消息通道")
+    private boolean autoPushChannel;
+
     public Integer getId() {
         return id;
     }
@@ -425,4 +428,12 @@ public class ParentPlatform {
     public void setAsMessageChannel(boolean asMessageChannel) {
         this.asMessageChannel = asMessageChannel;
     }
+
+    public boolean isAutoPushChannel() {
+        return autoPushChannel;
+    }
+
+    public void setAutoPushChannel(boolean autoPushChannel) {
+        this.autoPushChannel = autoPushChannel;
+    }
 }
@@ -32,11 +32,13 @@ public class SubscribeHolder {
     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 @@ public class SubscribeHolder {
         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) {
@@ -18,6 +18,9 @@ public class SubscribeInfo {
     }
+    public SubscribeInfo() {
+    }
+
     private String id;
     private SIPRequest request;
@@ -33,6 +36,21 @@ public class SubscribeInfo {
     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 class SubscribeInfo {
     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;
+    }
 }
@@ -93,7 +93,10 @@ public class CatalogEventLister implements ApplicationListener&lt;CatalogEvent&gt; {
                     }
                     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);
@@ -228,11 +228,11 @@ public class SIPRequestHeaderPlarformProvider {
 		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 @@ public class SIPRequestHeaderPlarformProvider {
 		// 设置编码， 防止中文乱码
 		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);
@@ -169,13 +169,13 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
         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();
     }
@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request;
 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 @@ public abstract class SIPRequestProcessorParent {
 				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的问题
@@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor
 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;
@@ -53,6 +54,10 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
 	@Autowired
 	private SIPSender sipSender;
+
+	@Autowired
+	private IPlatformService platformService;
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		// 添加消息处理的订阅
@@ -194,5 +199,8 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
 		} catch (SipException | InvalidArgumentException | ParseException e) {
 			logger.error("未处理的异常 ", e);
 		}
+		if (subscribeHolder.getCatalogSubscribe(platformId) == null && platform.isAutoPushChannel()) {
+			platformService.addSimulatedSubscribeInfo(platform);
+		}
 	}
 }
@@ -78,7 +78,6 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent 
         }
         taskExecutor.execute(()->{
             try {
-
                 String sn = getText(rootElement, "SN");
                 String channelId = getText(rootElement, "DeviceID");
                 RecordInfo recordInfo = new RecordInfo();
@@ -206,6 +206,13 @@ public class ZLMHttpHookListener {
         }
         // 推流鉴权的处理
         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) {
@@ -32,13 +32,20 @@ public class ZLMRESTfulUtils {
     }
     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 class ZLMRESTfulUtils {
     }
-
     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 @@ public class ZLMRESTfulUtils {
         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 @@ public class ZLMRESTfulUtils {
         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) {
@@ -41,6 +41,9 @@ public class StreamProxyItem extends GbStream {
     @Schema(description = "是否 无人观看时自动停用")
     private boolean enableDisableNoneReader;
+    @Schema(description = "拉流代理时zlm返回的key，用于停止拉流代理")
+    private String streamKey;
+
     public String getType() {
         return type;
     }
@@ -167,5 +170,11 @@ public class StreamProxyItem extends GbStream {
         this.enableAudio = enable_audio;
     }
+    public String getStreamKey() {
+        return streamKey;
+    }
+    public void setStreamKey(String streamKey) {
+        this.streamKey = streamKey;
+    }
 }
@@ -69,4 +69,6 @@ public interface IGbStreamService {
      * @param catalogId
      */
     void delAllPlatformInfo(String platformId, String catalogId);
+
+    List<GbStream> getGbChannelWithGbid(String gbId);
 }
@@ -80,4 +80,6 @@ public interface IPlatformService {
      * 语音喊话回复BYE
      */
     void stopBroadcast(ParentPlatform platform, DeviceChannel channel, String stream,boolean sendBye, MediaServerItem mediaServerItem);
+
+    void addSimulatedSubscribeInfo(ParentPlatform parentPlatform);
 }
@@ -18,6 +18,7 @@ import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 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 @@ public class GbStreamServiceImpl implements IGbStreamService {
             eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
         }
     }
+
+    @Override
+    public List<GbStream> getGbChannelWithGbid(String gbId) {
+        return gbStreamMapper.selectByGBId(gbId);
+    }
 }
@@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
 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.storager.dao.*;
 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 org.springframework.stereotype.Service;
 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 @@ public class PlatformServiceImpl implements IPlatformService {
             }
         }
+
         return false;
     }
@@ -243,19 +253,20 @@ public class PlatformServiceImpl implements IPlatformService {
                         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 @@ public class PlatformServiceImpl implements IPlatformService {
                     },
                     (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){
@@ -10,7 +10,10 @@ import com.genersoft.iot.vmp.conf.exception.ControllerException;
 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;
@@ -60,6 +63,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     private ZLMRESTfulUtils zlmresTfulUtils;
     @Autowired
+    private ZLMServerFactory zlmServerFactory;
+
+    @Autowired
     private StreamProxyMapper streamProxyMapper;
     @Autowired
@@ -137,7 +143,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
             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 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
             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 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
                     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 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         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 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
             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 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         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
@@ -440,7 +440,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
                 }
             }
-            if (streamPushItemListFroPlatform.size() > 0) {
+            if (!streamPushItemListFroPlatform.isEmpty()) {
                 platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
                 // 发送通知
                 for (String platformId : platformForEvent.keySet()) {
@@ -167,8 +167,8 @@ public interface DeviceChannelMapper {
             " <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" +
@@ -16,10 +16,10 @@ import java.util.List;
 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 @@ public interface ParentPlatformMapper {
             "ptz=#{ptz}, " +
             "rtcp=#{rtcp}, " +
             "as_message_channel=#{asMessageChannel}, " +
+            "auto_push_channel=#{autoPushChannel}, " +
             "status=#{status}, " +
             "start_offline_push=#{startOfflinePush}, " +
             "catalog_group=#{catalogGroup}, " +
@@ -12,9 +12,9 @@ import java.util.List;
 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 @@ public interface StreamProxyMapper {
             "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 @@ public interface StreamProxyMapper {
     @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")
@@ -36,6 +36,7 @@ import org.springframework.util.ObjectUtils;
 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 @@ public class DeviceQuery {
 		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);
 		}
 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.beans.factory.annotation.Autowired;
 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 @@ public class GbStreamController {
     private IGbStreamService gbStreamService;
     @Autowired
-    private IVideoManagerStorage storager;
+    private IPlatformService platformService;
     /**
@@ -107,4 +112,20 @@ public class GbStreamController {
             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);
+    }
 }
@@ -67,6 +67,16 @@ public class StreamProxyController {
         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 @@ public class StreamProxyController {
         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();
@@ -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 @@ export default {
         ptz: true,
         rtcp: false,
         asMessageChannel: false,
+        autoPushChannel: false,
         name: null,
         serverGBId: null,
         serverGBDomain: null,
@@ -208,6 +210,7 @@ export default {
         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 @@ export default {
         ptz: true,
         rtcp: false,
         asMessageChannel: false,
+        autoPushChannel: false,
         name: null,
         serverGBId: null,
         administrativeDivision: null,