From 470625e0770e27e8f20c3221b9c4096218079294 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 7 Aug 2023 17:00:01 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E5=85=A8=E5=B1=80=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E6=B5=81=E5=9C=B0=E5=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../request/impl/InviteRequestProcessor.java | 12 ++-- .../vmp/media/zlm/ZLMHttpHookListener.java | 18 +++++- .../zlm/dto/hook/HookResultForOnPublish.java | 10 ++++ .../iot/vmp/service/IInviteStreamService.java | 8 +++ .../service/impl/InviteStreamServiceImpl.java | 45 +++++++++++++-- .../iot/vmp/service/impl/PlayServiceImpl.java | 57 ++++--------------- 6 files changed, 88 insertions(+), 62 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 397a4a0f..d2d9657b 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -459,7 +459,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements sendRtpItem.setApp("rtp"); if ("Playback".equalsIgnoreCase(sessionName)) { sendRtpItem.setPlayType(InviteStreamType.PLAYBACK); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); + String startTimeStr = DateUtil.urlFormatter.format(start); + String endTimeStr = DateUtil.urlFormatter.format(end); + String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr; + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); sendRtpItem.setStreamId(ssrcInfo.getStream()); // 写入redis， 超时时回复 redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -520,12 +523,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } })); sendRtpItem.setPlayType(InviteStreamType.PLAY); - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - }else { - streamId = String.format("%08x", Integer.parseInt(ssrcInfo.getSsrc())).toUpperCase(); - } + String streamId = String.format("%s_%s", device.getDeviceId(), channelId); sendRtpItem.setStreamId(streamId); sendRtpItem.setSsrc(ssrcInfo.getSsrc()); redisCatchStorage.updateSendRTPSever(sendRtpItem); 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 465aa2f7..71bb5b86 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 @@ -193,7 +193,10 @@ public class ZLMHttpHookListener { String mediaServerId = json.getString("mediaServerId"); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); - + if (mediaInfo == null) { + return new HookResultForOnPublish(200, "success"); + } + // 推流鉴权的处理 if (!"rtp".equals(param.getApp())) { if (userSetting.getPushAuthority()) { // 推流鉴权 @@ -245,11 +248,21 @@ public class ZLMHttpHookListener { } }); + // 是否录像 if ("rtp".equals(param.getApp())) { result.setEnable_mp4(userSetting.getRecordSip()); } else { result.setEnable_mp4(userSetting.isRecordPushLive()); } + // 替换流地址 + if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) { + String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));; + InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc); + if (inviteInfo != null) { + result.setStream_replace(inviteInfo.getStream()); + logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream()); + } + } List ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream()); if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) { String deviceId = ssrcTransactionForAll.get(0).getDeviceId(); @@ -562,7 +575,7 @@ public class ZLMHttpHookListener { if ("rtp".equals(param.getApp())) { String[] s = param.getStream().split("_"); - if (!mediaInfo.isRtpEnable() || (s.length != 2 && s.length != 4)) { + if ((s.length != 2 && s.length != 4)) { defaultResult.setResult(HookResult.SUCCESS()); return defaultResult; } @@ -591,7 +604,6 @@ public class ZLMHttpHookListener { result.onTimeout(() -> { logger.info("[ZLM HOOK] 预览流自动点播, 等待超时"); - // 释放rtpserver msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时")); resultHolder.invokeResult(msg); }); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java index 68d969f4..12d83627 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java @@ -6,6 +6,7 @@ public class HookResultForOnPublish extends HookResult{ private boolean enable_mp4; private int mp4_max_second; private String mp4_save_path; + private String stream_replace; public HookResultForOnPublish() { } @@ -51,12 +52,21 @@ public class HookResultForOnPublish extends HookResult{ this.mp4_save_path = mp4_save_path; } + public String getStream_replace() { + return stream_replace; + } + + public void setStream_replace(String stream_replace) { + this.stream_replace = stream_replace; + } + @Override public String toString() { return "HookResultForOnPublish{" + "enable_audio=" + enable_audio + ", enable_mp4=" + enable_mp4 + ", mp4_max_second=" + mp4_max_second + + ", stream_replace=" + stream_replace + ", mp4_save_path='" + mp4_save_path + '\'' + '}'; } diff --git a/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java b/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java index ae30f26e..c06400dc 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java @@ -74,5 +74,13 @@ public interface IInviteStreamService { int getStreamInfoCount(String mediaServerId); + /** + * 获取MediaServer下的流信息 + */ + InviteInfo getInviteInfoBySSRC(String ssrc); + /** + * 更新ssrc + */ + InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrcInResponse); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java index def639b0..d630a2c0 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java @@ -80,7 +80,8 @@ public class InviteStreamServiceImpl implements IInviteStreamService { ":" + inviteInfoForUpdate.getType() + ":" + inviteInfoForUpdate.getDeviceId() + ":" + inviteInfoForUpdate.getChannelId() + - ":" + inviteInfoForUpdate.getStream(); + ":" + inviteInfoForUpdate.getStream()+ + ":" + inviteInfoForUpdate.getSsrcInfo().getSsrc(); redisTemplate.opsForValue().set(key, inviteInfoForUpdate); } @@ -96,7 +97,8 @@ public class InviteStreamServiceImpl implements IInviteStreamService { ":" + inviteInfo.getType() + ":" + inviteInfo.getDeviceId() + ":" + inviteInfo.getChannelId() + - ":" + stream; + ":" + stream + + ":" + inviteInfo.getSsrcInfo().getSsrc(); inviteInfoInDb.setStream(stream); if (inviteInfoInDb.getSsrcInfo() != null) { inviteInfoInDb.getSsrcInfo().setStream(stream); @@ -111,7 +113,8 @@ public class InviteStreamServiceImpl implements IInviteStreamService { ":" + (type != null ? type : "*") + ":" + (deviceId != null ? deviceId : "*") + ":" + (channelId != null ? channelId : "*") + - ":" + (stream != null ? stream : "*"); + ":" + (stream != null ? stream : "*") + + ":*"; List scanResult = RedisUtil.scan(redisTemplate, key); if (scanResult.size() != 1) { return null; @@ -136,7 +139,8 @@ public class InviteStreamServiceImpl implements IInviteStreamService { ":" + (type != null ? type : "*") + ":" + (deviceId != null ? deviceId : "*") + ":" + (channelId != null ? channelId : "*") + - ":" + (stream != null ? stream : "*"); + ":" + (stream != null ? stream : "*") + + ":*"; List scanResult = RedisUtil.scan(redisTemplate, scanKey); if (scanResult.size() > 0) { for (Object keyObj : scanResult) { @@ -191,7 +195,7 @@ public class InviteStreamServiceImpl implements IInviteStreamService { @Override public int getStreamInfoCount(String mediaServerId) { int count = 0; - String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*"; + String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:*"; List scanResult = RedisUtil.scan(redisTemplate, key); if (scanResult.size() == 0) { return 0; @@ -229,4 +233,35 @@ public class InviteStreamServiceImpl implements IInviteStreamService { } return key; } + + @Override + public InviteInfo getInviteInfoBySSRC(String ssrc) { + String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:" + ssrc; + List scanResult = RedisUtil.scan(redisTemplate, key); + if (scanResult.size() != 1) { + return null; + } + + return (InviteInfo) redisTemplate.opsForValue().get(scanResult.get(0)); + } + + @Override + public InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrc) { + InviteInfo inviteInfoInDb = getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); + if (inviteInfoInDb == null) { + return null; + } + removeInviteInfo(inviteInfoInDb); + String key = VideoManagerConstants.INVITE_PREFIX + + ":" + inviteInfo.getType() + + ":" + inviteInfo.getDeviceId() + + ":" + inviteInfo.getChannelId() + + ":" + inviteInfo.getStream() + + ":" + inviteInfo.getSsrcInfo().getSsrc(); + if (inviteInfoInDb.getSsrcInfo() != null) { + inviteInfoInDb.getSsrcInfo().setSsrc(ssrc); + } + redisTemplate.opsForValue().set(key, inviteInfoInDb); + return inviteInfoInDb; + } } 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 cf8bdd24..f2653f70 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 @@ -158,10 +158,7 @@ public class PlayServiceImpl implements IPlayService { } } } - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - } + String streamId = String.format("%s_%s", device.getDeviceId(), channelId);; SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); if (ssrcInfo == null) { callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null); @@ -457,16 +454,13 @@ public class PlayServiceImpl implements IPlayService { logger.warn("[录像回放] 单端口收流时不支持TCP主动方式收流 deviceId: {},channelId:{}", deviceId, channelId); throw new ControllerException(ErrorCode.ERROR100.getCode(), "单端口收流时不支持TCP主动方式收流"); } - String stream = null; - if (newMediaServerItem.isRtpEnable()) { - String startTimeStr = startTime.replace("-", "") - .replace(":", "") - .replace(" ", ""); - String endTimeTimeStr = endTime.replace("-", "") - .replace(":", "") - .replace(" ", ""); - stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr; - } + String startTimeStr = startTime.replace("-", "") + .replace(":", "") + .replace(" ", ""); + String endTimeTimeStr = endTime.replace("-", "") + .replace(":", "") + .replace(" ", ""); + String stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr; SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, callback); } @@ -628,44 +622,13 @@ public class PlayServiceImpl implements IPlayService { if (ssrcInResponse != null) { // 单端口 // 重新订阅流上线 - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", - ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); - subscribe.removeSubscribe(hookSubscribe); SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(inviteInfo.getDeviceId(), inviteInfo.getChannelId(), null, inviteInfo.getStream()); streamSession.remove(inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); - - String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase(); - hookSubscribe.getContent().put("stream", stream); - - inviteStreamService.updateInviteInfoForStream(inviteInfo, stream); + inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse); streamSession.put(device.getDeviceId(), channelId, ssrcTransaction.getCallId(), - stream, ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType); - subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> { - logger.info("[Invite 200OK] ssrc修正后收到订阅消息： " + hookParam); - dynamicTask.stop(timeOutTaskKey); - subscribe.removeSubscribe(hookSubscribe); - // hook响应 - StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInUse, hookParam, device.getDeviceId(), channelId); - if (streamInfo == null){ - callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); - inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); - return; - } - callback.run(InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), streamInfo); - inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null, - InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), - streamInfo); - if (inviteSessionType == InviteSessionType.PLAY) { - snapOnPlay(mediaServerItemInUse, device.getDeviceId(), channelId, stream); - } - }); + inviteInfo.getStream(), ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType); } } }