Commit d5e8aa62a11352f228ba449b204d53d4e17897a5

Authored by 648540858
1 parent 01344884

添加对海康平台录像回放的兼容,修复录像信息发送失败, 级联平台支持开启rtcp保活

src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java
@@ -113,7 +113,6 @@ public class ParentPlatform { @@ -113,7 +113,6 @@ public class ParentPlatform {
113 113
114 /** 114 /**
115 * RTCP流保活 115 * RTCP流保活
116 - * TODO 预留, 暂不实现  
117 */ 116 */
118 @Schema(description = "RTCP流保活") 117 @Schema(description = "RTCP流保活")
119 private boolean rtcp; 118 private boolean rtcp;
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
@@ -103,7 +103,7 @@ public interface ISIPCommander { @@ -103,7 +103,7 @@ public interface ISIPCommander {
103 * @param startTime 开始时间,格式要求:yyyy-MM-dd HH:mm:ss 103 * @param startTime 开始时间,格式要求:yyyy-MM-dd HH:mm:ss
104 * @param endTime 结束时间,格式要求:yyyy-MM-dd HH:mm:ss 104 * @param endTime 结束时间,格式要求:yyyy-MM-dd HH:mm:ss
105 */ 105 */
106 - void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event errorEvent); 106 + void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent);
107 107
108 /** 108 /**
109 * 请求历史媒体下载 109 * 请求历史媒体下载
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
@@ -456,7 +456,7 @@ public class SIPCommander implements ISIPCommander { @@ -456,7 +456,7 @@ public class SIPCommander implements ISIPCommander {
456 @Override 456 @Override
457 public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, 457 public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
458 String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent, 458 String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
459 - SipSubscribe.Event errorEvent) { 459 + SipSubscribe.Event okEvent,SipSubscribe.Event errorEvent) {
460 try { 460 try {
461 461
462 logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); 462 logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
@@ -535,10 +535,11 @@ public class SIPCommander implements ISIPCommander { @@ -535,10 +535,11 @@ public class SIPCommander implements ISIPCommander {
535 }); 535 });
536 Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc()); 536 Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
537 537
538 - transmitRequest(device, request, errorEvent, okEvent -> {  
539 - ResponseEvent responseEvent = (ResponseEvent) okEvent.event; 538 + transmitRequest(device, request, errorEvent, event -> {
  539 + ResponseEvent responseEvent = (ResponseEvent) event.event;
540 streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction(), VideoStreamSessionManager.SessionType.playback); 540 streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction(), VideoStreamSessionManager.SessionType.playback);
541 - streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog); 541 + streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), event.dialog);
  542 + okEvent.response(event);
542 }); 543 });
543 if (inviteStreamCallback != null) { 544 if (inviteStreamCallback != null) {
544 inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream())); 545 inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));
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 @@ -115,6 +115,11 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
115 param.put("pt", sendRtpItem.getPt()); 115 param.put("pt", sendRtpItem.getPt());
116 param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); 116 param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
117 param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); 117 param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
  118 + if (!sendRtpItem.isTcp() && parentPlatform.isRtcp()) {
  119 + // 开启rtcp保活
  120 + param.put("udp_rtcp_timeout", "1");
  121 + }
  122 +
118 if (mediaInfo == null) { 123 if (mediaInfo == null) {
119 RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( 124 RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
120 sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), 125 sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
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 @@ -98,8 +98,8 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
98 param.put("ssrc",sendRtpItem.getSsrc()); 98 param.put("ssrc",sendRtpItem.getSsrc());
99 logger.info("收到bye:停止向上级推流:" + streamId); 99 logger.info("收到bye:停止向上级推流:" + streamId);
100 MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); 100 MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
101 - zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);  
102 redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null); 101 redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
  102 + zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
103 int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId); 103 int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
104 if (totalReaderCount <= 0) { 104 if (totalReaderCount <= 0) {
105 logger.info("收到bye: {} 无其它观看者,通知设备停止推流", streamId); 105 logger.info("收到bye: {} 无其它观看者,通知设备停止推流", streamId);
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
@@ -563,6 +563,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements @@ -563,6 +563,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
563 responseAck(evt, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline"); 563 responseAck(evt, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
564 } else if ("push".equals(gbStream.getStreamType())) { 564 } else if ("push".equals(gbStream.getStreamType())) {
565 if (!platform.isStartOfflinePush()) { 565 if (!platform.isStartOfflinePush()) {
  566 + // 平台设置中关闭了拉起离线的推流则直接回复
566 responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel unavailable"); 567 responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel unavailable");
567 return; 568 return;
568 } 569 }
@@ -599,7 +600,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements @@ -599,7 +600,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
599 app, stream, channelId, mediaTransmissionTCP); 600 app, stream, channelId, mediaTransmissionTCP);
600 601
601 if (sendRtpItem == null) { 602 if (sendRtpItem == null) {
602 - logger.warn("服务器端口资源不足"); 603 + logger.warn("上级点时创建sendRTPItem失败,可能是服务器端口资源不足");
603 try { 604 try {
604 responseAck(evt, Response.BUSY_HERE); 605 responseAck(evt, Response.BUSY_HERE);
605 } catch (SipException e) { 606 } catch (SipException e) {
src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java
@@ -50,7 +50,7 @@ public class AssistRESTfulUtils { @@ -50,7 +50,7 @@ public class AssistRESTfulUtils {
50 if (mediaServerItem == null) { 50 if (mediaServerItem == null) {
51 return null; 51 return null;
52 } 52 }
53 - if (mediaServerItem.getRecordAssistPort() > 0) { 53 + if (mediaServerItem.getRecordAssistPort() <= 0) {
54 logger.warn("未启用Assist服务"); 54 logger.warn("未启用Assist服务");
55 return null; 55 return null;
56 } 56 }
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -19,8 +19,6 @@ import org.slf4j.Logger; @@ -19,8 +19,6 @@ import org.slf4j.Logger;
19 import org.slf4j.LoggerFactory; 19 import org.slf4j.LoggerFactory;
20 import org.springframework.beans.factory.annotation.Autowired; 20 import org.springframework.beans.factory.annotation.Autowired;
21 import org.springframework.beans.factory.annotation.Qualifier; 21 import org.springframework.beans.factory.annotation.Qualifier;
22 -import org.springframework.http.HttpStatus;  
23 -import org.springframework.http.ResponseEntity;  
24 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; 22 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
25 import org.springframework.util.ObjectUtils; 23 import org.springframework.util.ObjectUtils;
26 import org.springframework.web.bind.annotation.PostMapping; 24 import org.springframework.web.bind.annotation.PostMapping;
@@ -544,6 +542,8 @@ public class ZLMHttpHookListener { @@ -544,6 +542,8 @@ public class ZLMHttpHookListener {
544 for (SendRtpItem sendRtpItem : sendRtpItems) { 542 for (SendRtpItem sendRtpItem : sendRtpItems) {
545 ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); 543 ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
546 commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); 544 commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
  545 + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
  546 + sendRtpItem.getCallId(), sendRtpItem.getStreamId());
547 } 547 }
548 } 548 }
549 } 549 }
@@ -573,13 +573,19 @@ public class ZLMHttpHookListener { @@ -573,13 +573,19 @@ public class ZLMHttpHookListener {
573 return ret; 573 return ret;
574 }else { 574 }else {
575 StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId); 575 StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId);
576 - if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) {  
577 - ret.put("close", true);  
578 - streamProxyService.del(app, streamId);  
579 - String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();  
580 - logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除", app, streamId, url);  
581 - }else {  
582 - ret.put("close", false); 576 + if (streamProxyItem != null ) {
  577 + if (streamProxyItem.isEnable_remove_none_reader()) {
  578 + // 无人观看自动移除
  579 + ret.put("close", true);
  580 + streamProxyService.del(app, streamId);
  581 + String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();
  582 + logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除", app, streamId, url);
  583 + }else if (streamProxyItem.isEnable_disable_none_reader()) {
  584 + // 无人观看停用
  585 + ret.put("close", true);
  586 + }else {
  587 + ret.put("close", false);
  588 + }
583 } 589 }
584 return ret; 590 return ret;
585 } 591 }
@@ -626,7 +632,7 @@ public class ZLMHttpHookListener { @@ -626,7 +632,7 @@ public class ZLMHttpHookListener {
626 @ResponseBody 632 @ResponseBody
627 @PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8") 633 @PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8")
628 public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){ 634 public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
629 - 635 +
630 if (logger.isDebugEnabled()) { 636 if (logger.isDebugEnabled()) {
631 logger.debug("[ ZLM HOOK ]on_server_started API调用,参数:" + jsonObject.toString()); 637 logger.debug("[ ZLM HOOK ]on_server_started API调用,参数:" + jsonObject.toString());
632 } 638 }
@@ -649,6 +655,39 @@ public class ZLMHttpHookListener { @@ -649,6 +655,39 @@ public class ZLMHttpHookListener {
649 return ret; 655 return ret;
650 } 656 }
651 657
  658 + /**
  659 + * 发送rtp(startSendRtp)被动关闭时回调
  660 + */
  661 + @ResponseBody
  662 + @PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8")
  663 + public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody JSONObject jsonObject){
  664 +
  665 + logger.info("[ ZLM HOOK ]on_send_rtp_stopped API调用,参数:" + jsonObject);
  666 +
  667 + JSONObject ret = new JSONObject();
  668 + ret.put("code", 0);
  669 + ret.put("msg", "success");
  670 +
  671 + // 查找对应的上级推流,发送停止
  672 + String app = jsonObject.getString("app");
  673 + if (!"rtp".equals(app)) {
  674 + return ret;
  675 + }
  676 + String stream = jsonObject.getString("stream");
  677 + List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream);
  678 + if (sendRtpItems.size() > 0) {
  679 + for (SendRtpItem sendRtpItem : sendRtpItems) {
  680 + ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
  681 + commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
  682 + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
  683 + sendRtpItem.getCallId(), sendRtpItem.getStreamId());
  684 + }
  685 + }
  686 +
  687 +
  688 + return ret;
  689 + }
  690 +
652 private Map<String, String> urlParamToMap(String params) { 691 private Map<String, String> urlParamToMap(String params) {
653 HashMap<String, String> map = new HashMap<>(); 692 HashMap<String, String> map = new HashMap<>();
654 if (ObjectUtils.isEmpty(params)) { 693 if (ObjectUtils.isEmpty(params)) {
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java
@@ -37,6 +37,9 @@ public class StreamProxyItem extends GbStream { @@ -37,6 +37,9 @@ public class StreamProxyItem extends GbStream {
37 private boolean enable_mp4; 37 private boolean enable_mp4;
38 @Schema(description = "是否 无人观看时删除") 38 @Schema(description = "是否 无人观看时删除")
39 private boolean enable_remove_none_reader; 39 private boolean enable_remove_none_reader;
  40 +
  41 + @Schema(description = "是否 无人观看时不启用")
  42 + private boolean enable_disable_none_reader;
40 @Schema(description = "上级平台国标ID") 43 @Schema(description = "上级平台国标ID")
41 private String platformGbId; 44 private String platformGbId;
42 @Schema(description = "创建时间") 45 @Schema(description = "创建时间")
@@ -177,4 +180,11 @@ public class StreamProxyItem extends GbStream { @@ -177,4 +180,11 @@ public class StreamProxyItem extends GbStream {
177 this.enable_remove_none_reader = enable_remove_none_reader; 180 this.enable_remove_none_reader = enable_remove_none_reader;
178 } 181 }
179 182
  183 + public boolean isEnable_disable_none_reader() {
  184 + return enable_disable_none_reader;
  185 + }
  186 +
  187 + public void setEnable_disable_none_reader(boolean enable_disable_none_reader) {
  188 + this.enable_disable_none_reader = enable_disable_none_reader;
  189 + }
180 } 190 }
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
@@ -531,6 +531,7 @@ public class MediaServerServiceImpl implements IMediaServerService { @@ -531,6 +531,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
531 param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex)); 531 param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
532 param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex)); 532 param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
533 param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex)); 533 param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
  534 + param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex));
534 if (mediaServerItem.getRecordAssistPort() > 0) { 535 if (mediaServerItem.getRecordAssistPort() > 0) {
535 param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort())); 536 param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
536 }else { 537 }else {
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
@@ -73,7 +73,6 @@ public class MediaServiceImpl implements IMediaService { @@ -73,7 +73,6 @@ public class MediaServiceImpl implements IMediaService {
73 }else { 73 }else {
74 streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks, addr,null); 74 streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks, addr,null);
75 } 75 }
76 -  
77 } 76 }
78 } 77 }
79 return streamInfo; 78 return streamInfo;
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -291,7 +291,7 @@ public class PlayServiceImpl implements IPlayService { @@ -291,7 +291,7 @@ public class PlayServiceImpl implements IPlayService {
291 } 291 }
292 logger.info("[点播消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse ); 292 logger.info("[点播消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse );
293 if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { 293 if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
294 - logger.info("[SIP 消息] SSRC修正 {}->{}", ssrc, ssrcInResponse); 294 + logger.info("[点播消息] SSRC修正 {}->{}", ssrc, ssrcInResponse);
295 295
296 if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { 296 if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
297 // ssrc 不可用 297 // ssrc 不可用
@@ -441,37 +441,92 @@ public class PlayServiceImpl implements IPlayService { @@ -441,37 +441,92 @@ public class PlayServiceImpl implements IPlayService {
441 resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid); 441 resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid);
442 }, userSetting.getPlayTimeout()); 442 }, userSetting.getPlayTimeout());
443 443
  444 + SipSubscribe.Event errorEvent = event -> {
  445 + dynamicTask.stop(playBackTimeOutTaskKey);
  446 + requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg)));
  447 + playBackResult.setCode(ErrorCode.ERROR100.getCode());
  448 + playBackResult.setMsg(String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg));
  449 + playBackResult.setData(requestMessage);
  450 + playBackResult.setEvent(event);
  451 + playBackCallback.call(playBackResult);
  452 + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  453 + };
  454 +
  455 + InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
  456 + logger.info("收到回放订阅消息: " + inviteStreamInfo.getResponse().toJSONString());
  457 + dynamicTask.stop(playBackTimeOutTaskKey);
  458 + StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
  459 + if (streamInfo == null) {
  460 + logger.warn("设备回放API调用失败!");
  461 + playBackResult.setCode(ErrorCode.ERROR100.getCode());
  462 + playBackResult.setMsg("设备回放API调用失败!");
  463 + playBackCallback.call(playBackResult);
  464 + return;
  465 + }
  466 + redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
  467 + WVPResult<StreamInfo> success = WVPResult.success(streamInfo);
  468 + requestMessage.setData(success);
  469 + playBackResult.setCode(ErrorCode.SUCCESS.getCode());
  470 + playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
  471 + playBackResult.setData(requestMessage);
  472 + playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
  473 + playBackResult.setResponse(inviteStreamInfo.getResponse());
  474 + playBackCallback.call(playBackResult);
  475 + };
  476 +
444 cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack, 477 cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
445 - (InviteStreamInfo inviteStreamInfo) -> {  
446 - logger.info("收到订阅消息: " + inviteStreamInfo.getResponse().toJSONString());  
447 - dynamicTask.stop(playBackTimeOutTaskKey);  
448 - StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);  
449 - if (streamInfo == null) {  
450 - logger.warn("设备回放API调用失败!");  
451 - playBackResult.setCode(ErrorCode.ERROR100.getCode());  
452 - playBackResult.setMsg("设备回放API调用失败!");  
453 - playBackCallback.call(playBackResult);  
454 - return; 478 + hookEvent, eventResult -> {
  479 + if (eventResult.type == SipSubscribe.EventResultType.response) {
  480 + ResponseEvent responseEvent = (ResponseEvent)eventResult.event;
  481 + String contentString = new String(responseEvent.getResponse().getRawContent());
  482 + // 获取ssrc
  483 + int ssrcIndex = contentString.indexOf("y=");
  484 + // 检查是否有y字段
  485 + if (ssrcIndex >= 0) {
  486 + //ssrc规定长度为10字节,不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
  487 + String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
  488 + // 查询到ssrc不一致且开启了ssrc校验则需要针对处理
  489 + if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
  490 + return;
  491 + }
  492 + logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse );
  493 + if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
  494 + logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
  495 +
  496 + if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
  497 + // ssrc 不可用
  498 + // 释放ssrc
  499 + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  500 + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  501 + eventResult.msg = "下级自定义了ssrc,但是此ssrc不可用";
  502 + eventResult.statusCode = 400;
  503 + errorEvent.response(eventResult);
  504 + return;
  505 + }
  506 +
  507 + // 单端口模式streamId也有变化,需要重新设置监听
  508 + if (!mediaServerItem.isRtpEnable()) {
  509 + // 添加订阅
  510 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
  511 + subscribe.removeSubscribe(hookSubscribe);
  512 + hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
  513 + subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{
  514 + logger.info("[ZLM HOOK] ssrc修正后收到订阅消息: " + response.toJSONString());
  515 + dynamicTask.stop(playBackTimeOutTaskKey);
  516 + // hook响应
  517 + onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid);
  518 + hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
  519 + });
  520 + }
  521 + // 关闭rtp server
  522 + mediaServerService.closeRTPServer(device.getDeviceId(), channelId, ssrcInfo.getStream());
  523 + // 重新开启ssrc server
  524 + mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort());
  525 + }
  526 + }
455 } 527 }
456 - redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());  
457 - WVPResult<StreamInfo> success = WVPResult.success(streamInfo);  
458 - requestMessage.setData(success);  
459 - playBackResult.setCode(ErrorCode.SUCCESS.getCode());  
460 - playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());  
461 - playBackResult.setData(requestMessage);  
462 - playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());  
463 - playBackResult.setResponse(inviteStreamInfo.getResponse());  
464 - playBackCallback.call(playBackResult);  
465 - }, event -> {  
466 - dynamicTask.stop(playBackTimeOutTaskKey);  
467 - requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg)));  
468 - playBackResult.setCode(ErrorCode.ERROR100.getCode());  
469 - playBackResult.setMsg(String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg));  
470 - playBackResult.setData(requestMessage);  
471 - playBackResult.setEvent(event);  
472 - playBackCallback.call(playBackResult);  
473 - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());  
474 - }); 528 +
  529 + }, errorEvent);
475 return result; 530 return result;
476 } 531 }
477 532
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
@@ -236,4 +236,6 @@ public interface IRedisCatchStorage { @@ -236,4 +236,6 @@ public interface IRedisCatchStorage {
236 void sendStreamPushRequestedMsgForStatus(); 236 void sendStreamPushRequestedMsgForStatus();
237 237
238 List<SendRtpItem> querySendRTPServerByChnnelId(String channelId); 238 List<SendRtpItem> querySendRTPServerByChnnelId(String channelId);
  239 +
  240 + List<SendRtpItem> querySendRTPServerByStream(String stream);
239 } 241 }
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
@@ -388,6 +388,24 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { @@ -388,6 +388,24 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
388 } 388 }
389 389
390 @Override 390 @Override
  391 + public List<SendRtpItem> querySendRTPServerByStream(String stream) {
  392 + if (stream == null) {
  393 + return null;
  394 + }
  395 + String platformGbId = "*";
  396 + String callId = "*";
  397 + String channelId = "*";
  398 + String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + userSetting.getServerId() + "_" + platformGbId
  399 + + "_" + channelId + "_" + stream + "_" + callId;
  400 + List<Object> scan = RedisUtil.scan(key);
  401 + List<SendRtpItem> result = new ArrayList<>();
  402 + for (Object o : scan) {
  403 + result.add((SendRtpItem) RedisUtil.get((String) o));
  404 + }
  405 + return result;
  406 + }
  407 +
  408 + @Override
391 public List<SendRtpItem> querySendRTPServer(String platformGbId) { 409 public List<SendRtpItem> querySendRTPServer(String platformGbId) {
392 if (platformGbId == null) { 410 if (platformGbId == null) {
393 platformGbId = "*"; 411 platformGbId = "*";
web_src/src/components/dialog/platformEdit.vue
@@ -37,13 +37,13 @@ @@ -37,13 +37,13 @@
37 <el-form-item label="本地端口" prop="devicePort"> 37 <el-form-item label="本地端口" prop="devicePort">
38 <el-input v-model="platform.devicePort" :disabled="true" type="number"></el-input> 38 <el-input v-model="platform.devicePort" :disabled="true" type="number"></el-input>
39 </el-form-item> 39 </el-form-item>
  40 + <el-form-item label="SIP认证用户名" prop="username">
  41 + <el-input v-model="platform.username"></el-input>
  42 + </el-form-item>
40 </el-form> 43 </el-form>
41 </el-col> 44 </el-col>
42 <el-col :span="12"> 45 <el-col :span="12">
43 <el-form ref="platform2" :rules="rules" :model="platform" label-width="160px"> 46 <el-form ref="platform2" :rules="rules" :model="platform" label-width="160px">
44 - <el-form-item label="SIP认证用户名" prop="username">  
45 - <el-input v-model="platform.username"></el-input>  
46 - </el-form-item>  
47 <el-form-item label="行政区划" prop="administrativeDivision"> 47 <el-form-item label="行政区划" prop="administrativeDivision">
48 <el-input v-model="platform.administrativeDivision" clearable></el-input> 48 <el-input v-model="platform.administrativeDivision" clearable></el-input>
49 </el-form-item> 49 </el-form-item>
@@ -79,7 +79,7 @@ @@ -79,7 +79,7 @@
79 </el-select> 79 </el-select>
80 </el-form-item> 80 </el-form-item>
81 <el-form-item label="目录结构" prop="treeType" > 81 <el-form-item label="目录结构" prop="treeType" >
82 - <el-select v-model="platform.treeType" style="width: 100%" > 82 + <el-select v-model="platform.treeType" style="width: 100%" @change="treeTypeChange">
83 <el-option key="WGS84" label="行政区划" value="CivilCode"></el-option> 83 <el-option key="WGS84" label="行政区划" value="CivilCode"></el-option>
84 <el-option key="GCJ02" label="业务分组" value="BusinessGroup"></el-option> 84 <el-option key="GCJ02" label="业务分组" value="BusinessGroup"></el-option>
85 </el-select> 85 </el-select>
@@ -98,6 +98,7 @@ @@ -98,6 +98,7 @@
98 <el-checkbox label="启用" v-model="platform.enable" @change="checkExpires"></el-checkbox> 98 <el-checkbox label="启用" v-model="platform.enable" @change="checkExpires"></el-checkbox>
99 <el-checkbox label="云台控制" v-model="platform.ptz"></el-checkbox> 99 <el-checkbox label="云台控制" v-model="platform.ptz"></el-checkbox>
100 <el-checkbox label="拉起离线推流" v-model="platform.startOfflinePush"></el-checkbox> 100 <el-checkbox label="拉起离线推流" v-model="platform.startOfflinePush"></el-checkbox>
  101 + <el-checkbox label="RTCP保活" v-model="platform.rtcp" @change="rtcpCheckBoxChange"></el-checkbox>
101 </el-form-item> 102 </el-form-item>
102 <el-form-item> 103 <el-form-item>
103 <el-button type="primary" @click="onSubmit">{{ 104 <el-button type="primary" @click="onSubmit">{{
@@ -251,21 +252,7 @@ export default { @@ -251,21 +252,7 @@ export default {
251 252
252 }, 253 },
253 onSubmit: function () { 254 onSubmit: function () {
254 - if (this.onSubmit_text === "保存") {  
255 - this.$confirm("修改目录结构会导致关联目录与通道数据被清空", '提示', {  
256 - dangerouslyUseHTMLString: true,  
257 - confirmButtonText: '确定',  
258 - cancelButtonText: '取消',  
259 - center: true,  
260 - type: 'warning'  
261 - }).then(() => {  
262 - this.saveForm()  
263 - }).catch(() => {  
264 -  
265 - });  
266 - }else {  
267 - this.saveForm()  
268 - } 255 + this.saveForm()
269 }, 256 },
270 saveForm: function (){ 257 saveForm: function (){
271 this.$axios({ 258 this.$axios({
@@ -343,6 +330,22 @@ export default { @@ -343,6 +330,22 @@ export default {
343 if (this.platform.enable && this.platform.expires == "0") { 330 if (this.platform.enable && this.platform.expires == "0") {
344 this.platform.expires = "300"; 331 this.platform.expires = "300";
345 } 332 }
  333 + },
  334 + rtcpCheckBoxChange: function (result){
  335 + if (result) {
  336 + this.$message({
  337 + showClose: true,
  338 + message: "开启RTCP保活需要上级平台支持,可以避免无效推流",
  339 + type: "warning",
  340 + });
  341 + }
  342 + },
  343 + treeTypeChange: function (){
  344 + this.$message({
  345 + showClose: true,
  346 + message: "修改目录结构会导致关联目录与通道数据被清空,保存后生效",
  347 + type: "warning",
  348 + });
346 } 349 }
347 }, 350 },
348 }; 351 };