From d5e8aa62a11352f228ba449b204d53d4e17897a5 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 5 Sep 2022 17:10:21 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=AF=B9=E6=B5=B7=E5=BA=B7?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E5=BD=95=E5=83=8F=E5=9B=9E=E6=94=BE=E7=9A=84?= =?UTF-8?q?=E5=85=BC=E5=AE=B9=EF=BC=8C=E4=BF=AE=E5=A4=8D=E5=BD=95=E5=83=8F?= =?UTF-8?q?=E4=BF=A1=E6=81=AF=E5=8F=91=E9=80=81=E5=A4=B1=E8=B4=A5,=20?= =?UTF-8?q?=E7=BA=A7=E8=81=94=E5=B9=B3=E5=8F=B0=E6=94=AF=E6=8C=81=E5=BC=80?= =?UTF-8?q?=E5=90=AFrtcp=E4=BF=9D=E6=B4=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/gb28181/bean/ParentPlatform.java | 1 - .../gb28181/transmit/cmd/ISIPCommander.java | 2 +- .../transmit/cmd/impl/SIPCommander.java | 9 +- .../request/impl/AckRequestProcessor.java | 5 + .../request/impl/ByeRequestProcessor.java | 2 +- .../request/impl/InviteRequestProcessor.java | 3 +- .../iot/vmp/media/zlm/AssistRESTfulUtils.java | 2 +- .../vmp/media/zlm/ZLMHttpHookListener.java | 59 +++++++-- .../vmp/media/zlm/dto/StreamProxyItem.java | 10 ++ .../service/impl/MediaServerServiceImpl.java | 1 + .../vmp/service/impl/MediaServiceImpl.java | 1 - .../iot/vmp/service/impl/PlayServiceImpl.java | 115 +++++++++++++----- .../iot/vmp/storager/IRedisCatchStorage.java | 2 + .../storager/impl/RedisCatchStorageImpl.java | 18 +++ .../src/components/dialog/platformEdit.vue | 41 ++++--- 15 files changed, 202 insertions(+), 69 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java index 00a16f99..c8ab2e84 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java @@ -113,7 +113,6 @@ public class ParentPlatform { /** * RTCP流保活 - * TODO 预留, 暂不实现 */ @Schema(description = "RTCP流保活") private boolean rtcp; diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java index 068d2dfc..8aadf2c8 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java @@ -103,7 +103,7 @@ public interface ISIPCommander { * @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss * @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss */ - void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event errorEvent); + void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent); /** * 请求历史媒体下载 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index f006eccd..4d1e568b 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -456,7 +456,7 @@ public class SIPCommander implements ISIPCommander { @Override public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent, - SipSubscribe.Event errorEvent) { + SipSubscribe.Event okEvent,SipSubscribe.Event errorEvent) { try { logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); @@ -535,10 +535,11 @@ public class SIPCommander implements ISIPCommander { }); Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc()); - transmitRequest(device, request, errorEvent, okEvent -> { - ResponseEvent responseEvent = (ResponseEvent) okEvent.event; + transmitRequest(device, request, errorEvent, event -> { + ResponseEvent responseEvent = (ResponseEvent) event.event; streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction(), VideoStreamSessionManager.SessionType.playback); - streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog); + streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), event.dialog); + okEvent.response(event); }); if (inviteStreamCallback != null) { inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream())); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java index c46f181d..a56a83c2 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java @@ -115,6 +115,11 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In param.put("pt", sendRtpItem.getPt()); param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); + if (!sendRtpItem.isTcp() && parentPlatform.isRtcp()) { + // 开启rtcp保活 + param.put("udp_rtcp_timeout", "1"); + } + if (mediaInfo == null) { RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java index 88973769..706d4229 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java @@ -98,8 +98,8 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In param.put("ssrc",sendRtpItem.getSsrc()); logger.info("收到bye:停止向上级推流：" + streamId); MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param); redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null); + zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param); int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId); if (totalReaderCount <= 0) { logger.info("收到bye: {} 无其它观看者，通知设备停止推流", streamId); 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 5286aa51..a6956dab 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 @@ -563,6 +563,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements responseAck(evt, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline"); } else if ("push".equals(gbStream.getStreamType())) { if (!platform.isStartOfflinePush()) { + // 平台设置中关闭了拉起离线的推流则直接回复 responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel unavailable"); return; } @@ -599,7 +600,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements app, stream, channelId, mediaTransmissionTCP); if (sendRtpItem == null) { - logger.warn("服务器端口资源不足"); + logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足"); try { responseAck(evt, Response.BUSY_HERE); } catch (SipException e) { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java index 2d117543..345ea1e7 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java @@ -50,7 +50,7 @@ public class AssistRESTfulUtils { if (mediaServerItem == null) { return null; } - if (mediaServerItem.getRecordAssistPort() > 0) { + if (mediaServerItem.getRecordAssistPort() <= 0) { logger.warn("未启用Assist服务"); return null; } 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 78b59d90..b7899742 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 @@ -19,8 +19,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.ObjectUtils; import org.springframework.web.bind.annotation.PostMapping; @@ -544,6 +542,8 @@ public class ZLMHttpHookListener { for (SendRtpItem sendRtpItem : sendRtpItems) { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); } } } @@ -573,13 +573,19 @@ public class ZLMHttpHookListener { return ret; }else { StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId); - if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) { - ret.put("close", true); - streamProxyService.del(app, streamId); - String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url(); - logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除", app, streamId, url); - }else { - ret.put("close", false); + if (streamProxyItem != null ) { + if (streamProxyItem.isEnable_remove_none_reader()) { + // 无人观看自动移除 + ret.put("close", true); + streamProxyService.del(app, streamId); + String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url(); + logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除", app, streamId, url); + }else if (streamProxyItem.isEnable_disable_none_reader()) { + // 无人观看停用 + ret.put("close", true); + }else { + ret.put("close", false); + } } return ret; } @@ -626,7 +632,7 @@ public class ZLMHttpHookListener { @ResponseBody @PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8") public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){ - + if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_server_started API调用，参数：" + jsonObject.toString()); } @@ -649,6 +655,39 @@ public class ZLMHttpHookListener { return ret; } + /** + * 发送rtp(startSendRtp)被动关闭时回调 + */ + @ResponseBody + @PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8") + public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody JSONObject jsonObject){ + + logger.info("[ ZLM HOOK ]on_send_rtp_stopped API调用，参数：" + jsonObject); + + JSONObject ret = new JSONObject(); + ret.put("code", 0); + ret.put("msg", "success"); + + // 查找对应的上级推流，发送停止 + String app = jsonObject.getString("app"); + if (!"rtp".equals(app)) { + return ret; + } + String stream = jsonObject.getString("stream"); + List sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream); + if (sendRtpItems.size() > 0) { + for (SendRtpItem sendRtpItem : sendRtpItems) { + ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); + commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); + } + } + + + return ret; + } + private Map urlParamToMap(String params) { HashMap map = new HashMap<>(); if (ObjectUtils.isEmpty(params)) { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java index 897e9e3a..f0d08a14 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java @@ -37,6 +37,9 @@ public class StreamProxyItem extends GbStream { private boolean enable_mp4; @Schema(description = "是否 无人观看时删除") private boolean enable_remove_none_reader; + + @Schema(description = "是否 无人观看时不启用") + private boolean enable_disable_none_reader; @Schema(description = "上级平台国标ID") private String platformGbId; @Schema(description = "创建时间") @@ -177,4 +180,11 @@ public class StreamProxyItem extends GbStream { this.enable_remove_none_reader = enable_remove_none_reader; } + public boolean isEnable_disable_none_reader() { + return enable_disable_none_reader; + } + + public void setEnable_disable_none_reader(boolean enable_disable_none_reader) { + this.enable_disable_none_reader = enable_disable_none_reader; + } } 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 37aeca0d..64a411aa 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 @@ -531,6 +531,7 @@ public class MediaServerServiceImpl implements IMediaServerService { 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.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex)); + param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex)); if (mediaServerItem.getRecordAssistPort() > 0) { param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort())); }else { diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java index 6288a164..dd3800f2 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java @@ -73,7 +73,6 @@ public class MediaServiceImpl implements IMediaService { }else { streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks, addr,null); } - } } return streamInfo; 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 5ccc5e83..aa019229 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 @@ -291,7 +291,7 @@ public class PlayServiceImpl implements IPlayService { } logger.info("[点播消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse ); if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { - logger.info("[SIP 消息] SSRC修正 {}->{}", ssrc, ssrcInResponse); + logger.info("[点播消息] SSRC修正 {}->{}", ssrc, ssrcInResponse); if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { // ssrc 不可用 @@ -441,37 +441,92 @@ public class PlayServiceImpl implements IPlayService { resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid); }, userSetting.getPlayTimeout()); + SipSubscribe.Event errorEvent = event -> { + dynamicTask.stop(playBackTimeOutTaskKey); + requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg))); + playBackResult.setCode(ErrorCode.ERROR100.getCode()); + playBackResult.setMsg(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg)); + playBackResult.setData(requestMessage); + playBackResult.setEvent(event); + playBackCallback.call(playBackResult); + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + }; + + InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> { + logger.info("收到回放订阅消息： " + inviteStreamInfo.getResponse().toJSONString()); + dynamicTask.stop(playBackTimeOutTaskKey); + StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); + if (streamInfo == null) { + logger.warn("设备回放API调用失败！"); + playBackResult.setCode(ErrorCode.ERROR100.getCode()); + playBackResult.setMsg("设备回放API调用失败！"); + playBackCallback.call(playBackResult); + return; + } + redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId()); + WVPResult success = WVPResult.success(streamInfo); + requestMessage.setData(success); + playBackResult.setCode(ErrorCode.SUCCESS.getCode()); + playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); + playBackResult.setData(requestMessage); + playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); + playBackResult.setResponse(inviteStreamInfo.getResponse()); + playBackCallback.call(playBackResult); + }; + cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack, - (InviteStreamInfo inviteStreamInfo) -> { - logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString()); - dynamicTask.stop(playBackTimeOutTaskKey); - StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); - if (streamInfo == null) { - logger.warn("设备回放API调用失败！"); - playBackResult.setCode(ErrorCode.ERROR100.getCode()); - playBackResult.setMsg("设备回放API调用失败！"); - playBackCallback.call(playBackResult); - return; + hookEvent, eventResult -> { + if (eventResult.type == SipSubscribe.EventResultType.response) { + ResponseEvent responseEvent = (ResponseEvent)eventResult.event; + String contentString = new String(responseEvent.getResponse().getRawContent()); + // 获取ssrc + int ssrcIndex = contentString.indexOf("y="); + // 检查是否有y字段 + if (ssrcIndex >= 0) { + //ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容 + String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); + // 查询到ssrc不一致且开启了ssrc校验则需要针对处理 + if (ssrcInfo.getSsrc().equals(ssrcInResponse)) { + return; + } + logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse ); + if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { + logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); + + if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { + // ssrc 不可用 + // 释放ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + eventResult.msg = "下级自定义了ssrc,但是此ssrc不可用"; + eventResult.statusCode = 400; + errorEvent.response(eventResult); + return; + } + + // 单端口模式streamId也有变化，需要重新设置监听 + if (!mediaServerItem.isRtpEnable()) { + // 添加订阅 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); + subscribe.removeSubscribe(hookSubscribe); + hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); + subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{ + logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString()); + dynamicTask.stop(playBackTimeOutTaskKey); + // hook响应 + onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); + hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream())); + }); + } + // 关闭rtp server + mediaServerService.closeRTPServer(device.getDeviceId(), channelId, ssrcInfo.getStream()); + // 重新开启ssrc server + mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort()); + } + } } - redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId()); - WVPResult success = WVPResult.success(streamInfo); - requestMessage.setData(success); - playBackResult.setCode(ErrorCode.SUCCESS.getCode()); - playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); - playBackResult.setData(requestMessage); - playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); - playBackResult.setResponse(inviteStreamInfo.getResponse()); - playBackCallback.call(playBackResult); - }, event -> { - dynamicTask.stop(playBackTimeOutTaskKey); - requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg))); - playBackResult.setCode(ErrorCode.ERROR100.getCode()); - playBackResult.setMsg(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg)); - playBackResult.setData(requestMessage); - playBackResult.setEvent(event); - playBackCallback.call(playBackResult); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - }); + + }, errorEvent); return result; } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 1f467e4e..f66b3011 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -236,4 +236,6 @@ public interface IRedisCatchStorage { void sendStreamPushRequestedMsgForStatus(); List querySendRTPServerByChnnelId(String channelId); + + List querySendRTPServerByStream(String stream); } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index ecefe737..9d30fef3 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -387,6 +387,24 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { return result; } + @Override + public List querySendRTPServerByStream(String stream) { + if (stream == null) { + return null; + } + String platformGbId = "*"; + String callId = "*"; + String channelId = "*"; + String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + userSetting.getServerId() + "_" + platformGbId + + "_" + channelId + "_" + stream + "_" + callId; + List scan = RedisUtil.scan(key); + List result = new ArrayList<>(); + for (Object o : scan) { + result.add((SendRtpItem) RedisUtil.get((String) o)); + } + return result; + } + @Override public List querySendRTPServer(String platformGbId) { if (platformGbId == null) { diff --git a/web_src/src/components/dialog/platformEdit.vue b/web_src/src/components/dialog/platformEdit.vue index ef28b4c7..76382324 100644 --- a/web_src/src/components/dialog/platformEdit.vue +++ b/web_src/src/components/dialog/platformEdit.vue @@ -37,13 +37,13 @@ + + + - - - @@ -79,7 +79,7 @@ - + @@ -98,6 +98,7 @@ + {{ @@ -251,21 +252,7 @@ export default { }, onSubmit: function () { - if (this.onSubmit_text === "保存") { - this.$confirm("修改目录结构会导致关联目录与通道数据被清空", '提示', { - dangerouslyUseHTMLString: true, - confirmButtonText: '确定', - cancelButtonText: '取消', - center: true, - type: 'warning' - }).then(() => { - this.saveForm() - }).catch(() => { - - }); - }else { - this.saveForm() - } + this.saveForm() }, saveForm: function (){ this.$axios({ @@ -343,6 +330,22 @@ export default { if (this.platform.enable && this.platform.expires == "0") { this.platform.expires = "300"; } + }, + rtcpCheckBoxChange: function (result){ + if (result) { + this.$message({ + showClose: true, + message: "开启RTCP保活需要上级平台支持，可以避免无效推流", + type: "warning", + }); + } + }, + treeTypeChange: function (){ + this.$message({ + showClose: true, + message: "修改目录结构会导致关联目录与通道数据被清空，保存后生效", + type: "warning", + }); } }, };