From 95ed66293532268a26970d3c06795641999e0712 Mon Sep 17 00:00:00 2001 From: 64850858 <648540858@qq.com> Date: Mon, 26 Jul 2021 17:22:40 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dstream-none-reader-delay-ms?= =?UTF-8?q?=E4=B8=BA-1=E6=97=B6=E4=B8=8D=E8=87=AA=E5=8A=A8=E5=85=B3?= =?UTF-8?q?=E9=97=AD=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../gb28181/transmit/SIPProcessorFactory.java | 1 + .../request/impl/ByeRequestProcessor.java | 53 ++++++++++++++----- .../request/impl/MessageRequestProcessor.java | 1 - .../response/impl/ByeResponseProcessor.java | 2 +- .../vmp/media/zlm/ZLMHttpHookListener.java | 8 ++- .../service/impl/MediaServerServiceImpl.java | 4 +- .../iot/vmp/service/impl/PlayServiceImpl.java | 4 +- .../vmp/storager/IVideoManagerStorager.java | 2 + .../vmp/storager/dao/DeviceChannelMapper.java | 3 ++ .../impl/VideoManagerStoragerImpl.java | 10 ++++ .../vmanager/gb28181/play/PlayController.java | 5 +- 11 files changed, 70 insertions(+), 23 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java index 30c9e95c..6a93d013 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java @@ -159,6 +159,7 @@ public class SIPProcessorFactory { ByeRequestProcessor processor = new ByeRequestProcessor(); processor.setRequestEvent(evt); processor.setRedisCatchStorage(redisCatchStorage); + processor.setStorager(storager); processor.setZlmrtpServerFactory(zlmrtpServerFactory); processor.setSIPCommander(cmder); processor.setMediaServerService(mediaServerService); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java index ffa53778..817b6b7b 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java @@ -11,6 +11,8 @@ import javax.sip.header.HeaderAddress; import javax.sip.header.ToHeader; import javax.sip.message.Response; +import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.request.SIPRequestAbstractProcessor; @@ -18,6 +20,7 @@ import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,6 +41,8 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor { private IRedisCatchStorage redisCatchStorage; + private IVideoManagerStorager storager; + private ZLMRTPServerFactory zlmrtpServerFactory; private IMediaServerService mediaServerService; @@ -56,20 +61,32 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor { String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId); - if (sendRtpItem == null) return; - String streamId = sendRtpItem.getStreamId(); - Map param = new HashMap<>(); - param.put("vhost","__defaultVhost__"); - param.put("app",sendRtpItem.getApp()); - param.put("stream",streamId); - param.put("ssrc",sendRtpItem.getSsrc()); - logger.info("停止向上级推流：" + streamId); - MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param); - redisCatchStorage.deleteSendRTPServer(platformGbId, channelId); - if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) { - logger.info(streamId + "无其它观看者，通知设备停止推流"); - cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId); + logger.info("收到bye, [{}/{}]", platformGbId, channelId); + if (sendRtpItem != null){ + String streamId = sendRtpItem.getStreamId(); + Map param = new HashMap<>(); + param.put("vhost","__defaultVhost__"); + param.put("app",sendRtpItem.getApp()); + param.put("stream",streamId); + param.put("ssrc",sendRtpItem.getSsrc()); + logger.info("停止向上级推流：" + streamId); + MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param); + redisCatchStorage.deleteSendRTPServer(platformGbId, channelId); + if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) { + logger.info(streamId + "无其它观看者，通知设备停止推流"); + cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId); + } + } + // 可能是设备主动停止 + Device device = storager.queryVideoDeviceByChannelId(platformGbId); + if (device != null) { + StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId); + if (streamInfo != null) { + redisCatchStorage.stopPlay(streamInfo); + } + storager.stopPlay(device.getDeviceId(), channelId); + mediaServerService.closeRTPServer(device, channelId); } } } catch (SipException e) { @@ -124,4 +141,12 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor { public void setMediaServerService(IMediaServerService mediaServerService) { this.mediaServerService = mediaServerService; } + + public IVideoManagerStorager getStorager() { + return storager; + } + + public void setStorager(IVideoManagerStorager storager) { + this.storager = storager; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java index d46c0c40..6a29652c 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java @@ -756,7 +756,6 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor { deferredResultHolder.invokeResult(msg); } } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { - // } catch (DocumentException e) { e.printStackTrace(); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/ByeResponseProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/ByeResponseProcessor.java index 53bc8f79..02fdb1ae 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/ByeResponseProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/ByeResponseProcessor.java @@ -27,7 +27,7 @@ public class ByeResponseProcessor implements ISIPResponseProcessor { @Override public void process(ResponseEvent evt, SipLayer layer, SipConfig config) { // TODO Auto-generated method stub - + System.out.println( ); } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index 478772aa..678a56e2 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -346,6 +346,10 @@ public class ZLMHttpHookListener { redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch); } } + MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); + if (mediaServerItem != null && "-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())) { + ret.put("close", false); + } return new ResponseEntity(ret.toString(),HttpStatus.OK); }else { JSONObject ret = new JSONObject(); @@ -371,7 +375,7 @@ public class ZLMHttpHookListener { if (userSetup.isAutoApplyPlay() && mediaInfo != null) { String app = json.getString("app"); String streamId = json.getString("stream"); - if ("rtp".equals(app) && streamId.contains("gb_play") ) { + if ("rtp".equals(app)) { String[] s = streamId.split("_"); if (s.length == 4) { String deviceId = s[2]; @@ -382,7 +386,7 @@ public class ZLMHttpHookListener { SSRCInfo ssrcInfo; String streamId2 = null; if (mediaInfo.isRtpEnable()) { - streamId2 = String.format("gb_play_%s_%s", device.getDeviceId(), channelId); + streamId2 = String.format("%s/%s", device.getDeviceId(), channelId); } ssrcInfo = mediaServerService.openRTPServer(mediaInfo, streamId2); cmder.playStreamCmd(mediaInfo, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index 3d666658..ba428b59 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -123,7 +123,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId); MediaServerItem mediaServerItem = this.getOne(mediaServerId); if (mediaServerItem != null) { - String streamId = String.format("gb_play_%s_%s", device.getDeviceId(), channelId); + String streamId = String.format("%s/%s", device.getDeviceId(), channelId); zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId); releaseSsrc(mediaServerItem, streamSession.getSSRC(device.getDeviceId(), channelId)); } @@ -380,7 +380,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex)); param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex)); param.put("hook.timeoutSec","20"); - param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()); + param.put("general.streamNoneReaderDelayMS","-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() ); JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param); diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index c21f599b..eb80e784 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -156,7 +156,7 @@ public class PlayServiceImpl implements IPlayService { SSRCInfo ssrcInfo; String streamId = null; if (mediaServerItem.isRtpEnable()) { - streamId = String.format("gb_play_%s_%s", device.getDeviceId(), channelId); + streamId = String.format("%s/%s", device.getDeviceId(), channelId); } ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId); @@ -221,7 +221,7 @@ public class PlayServiceImpl implements IPlayService { SSRCInfo ssrcInfo; String streamId2 = null; if (mediaServerItem.isRtpEnable()) { - streamId2 = String.format("gb_play_%s_%s", device.getDeviceId(), channelId); + streamId2 = String.format("%s/%s", device.getDeviceId(), channelId); } ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java b/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java index 223a5e88..266c9455 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java @@ -374,4 +374,6 @@ public interface IVideoManagerStorager { void updateMediaServer(MediaServerItem mediaServerItem); List getStreamProxyListForEnableInMediaServer(String id, boolean b); + + Device queryVideoDeviceByChannelId(String platformGbId); } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java index 45b5ab7b..f7b89a98 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java @@ -102,4 +102,7 @@ public interface DeviceChannelMapper { " "}) List queryChannelListInAll(String query, Boolean online, Boolean hasSubChannel, String platformId, Boolean inPlatform); + + @Select("SELECT * FROM device_channel WHERE channelId=#{channelId}") + List queryChannelByChannelId(String channelId); } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java index 4e247531..82d9bdb5 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java @@ -623,4 +623,14 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager { public List getStreamProxyListForEnableInMediaServer(String id, boolean enable) { return streamProxyMapper.selectForEnableInMediaServer(id, enable); } + + @Override + public Device queryVideoDeviceByChannelId(String channelId) { + Device result = null; + List channelList = deviceChannelMapper.queryChannelByChannelId(channelId); + if (channelList.size() == 1) { + result = deviceMapper.getDeviceByDeviceId(channelList.get(0).getDeviceId()); + } + return result; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java index d6545b5f..a5a94a9d 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java @@ -168,7 +168,10 @@ public class PlayController { }) @PostMapping("/convert/{streamId}") public ResponseEntity playConvert(@PathVariable String streamId) { - StreamInfo streamInfo = streamId.startsWith("gb_play_") ? redisCatchStorage.queryPlayByStreamId(streamId) : redisCatchStorage.queryPlaybackByStreamId(streamId); + StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId); + if (streamInfo == null) { + streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId); + } if (streamInfo == null) { logger.warn("视频转码API调用失败！, 视频流已经停止!"); return new ResponseEntity("未找到视频流信息, 视频流可能已经停止", HttpStatus.OK);