From 483d5e04a71b6632419699df1077ac5a5457ad38 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 25 Sep 2023 18:51:02 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=8B=89=E6=B5=81=E4=BB=A3?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../subscribe/catalog/CatalogEventLister.java | 2 +- .../iot/vmp/media/zlm/ZLMRESTfulUtils.java | 21 ++++++++++++---- .../service/impl/StreamProxyServiceImpl.java | 24 ++++++++++++------- .../streamProxy/StreamProxyController.java | 3 +++ 4 files changed, 36 insertions(+), 14 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java index 89ecb186..122bc545 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java @@ -93,7 +93,7 @@ public class CatalogEventLister implements ApplicationListener { } 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().equals("push") && !userSetting.isUsePushingAsStatus()) { continue; } DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java index e58f2aeb..0403eecf 100755 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java @@ -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 param, RequestCallback callback) { - OkHttpClient client = getClient(); + return sendPost(mediaServerItem, api, param, callback, null); + } + + + public JSONObject sendPost(MediaServerItem mediaServerItem, String api, Map param, RequestCallback callback, Integer readTimeOut) { + OkHttpClient client = getClient(readTimeOut); if (mediaServerItem == null) { return null; @@ -310,7 +321,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) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java index a4e00ad4..f36cff90 100755 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java @@ -11,6 +11,8 @@ 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.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; @@ -160,15 +162,20 @@ 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); +// 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); +// } +// }, 3000); String delayTalkKey = UUID.randomUUID().toString(); dynamicTask.startDelay(delayTalkKey, ()->{ StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaInfo.getId(), false); @@ -178,9 +185,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); diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/streamProxy/StreamProxyController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/streamProxy/StreamProxyController.java index 0689f423..fe6df729 100755 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/streamProxy/StreamProxyController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/streamProxy/StreamProxyController.java @@ -80,6 +80,9 @@ 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); }