From 013448842833ebed52ef13846572408b7095bf7b Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: Fri, 2 Sep 2022 17:00:41 +0800
Subject: [PATCH 1/5] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=8E=A8=E6=B5=81?=
=?UTF-8?q?=E5=88=97=E8=A1=A8=E6=9F=A5=E8=AF=A2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../transmit/event/request/impl/SubscribeRequestProcessor.java | 2 --
.../com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java | 2 +-
2 files changed, 1 insertion(+), 3 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java
index 4b7629ce9..8c04368a1 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java
@@ -211,7 +211,6 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
}else if (subscribeInfo.getExpires() == 0) {
subscribeHolder.removeCatalogSubscribe(platformId);
}
-
try {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
responseXmlAck(evt, resultXml.toString(), parentPlatform);
@@ -219,5 +218,4 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
e.printStackTrace();
}
}
-
}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
index abff3eb97..0e656cffe 100644
--- a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
@@ -77,7 +77,7 @@ public interface StreamPushMapper {
"1=1 " +
" AND (st.app LIKE '%${query}%' OR st.stream LIKE '%${query}%' OR gs.gbId LIKE '%${query}%' OR gs.name LIKE '%${query}%') " +
" AND (gs.gbId is null OR st.pushIng=1)" +
- " AND st.pushIng=0" +
+ " AND (gs.pushIng is null OR st.pushIng=0) " +
" AND st.mediaServerId=#{mediaServerId} " +
"order by st.createTime desc" +
" "})
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 2/5] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=AF=B9=E6=B5=B7?=
=?UTF-8?q?=E5=BA=B7=E5=B9=B3=E5=8F=B0=E5=BD=95=E5=83=8F=E5=9B=9E=E6=94=BE?=
=?UTF-8?q?=E7=9A=84=E5=85=BC=E5=AE=B9=EF=BC=8C=E4=BF=AE=E5=A4=8D=E5=BD=95?=
=?UTF-8?q?=E5=83=8F=E4=BF=A1=E6=81=AF=E5=8F=91=E9=80=81=E5=A4=B1=E8=B4=A5?=
=?UTF-8?q?,=20=E7=BA=A7=E8=81=94=E5=B9=B3=E5=8F=B0=E6=94=AF=E6=8C=81?=
=?UTF-8?q?=E5=BC=80=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 00a16f99f..c8ab2e849 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 068d2dfc9..8aadf2c8a 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 f006eccd5..4d1e568bc 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 c46f181d2..a56a83c21 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 889737696..706d42297 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 5286aa511..a6956dab5 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 2d117543f..345ea1e70 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 78b59d90c..b78997427 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 897e9e3a0..f0d08a145 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 37aeca0d6..64a411aa0 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 6288a1645..dd3800f2a 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 5ccc5e83c..aa019229e 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 1f467e4eb..f66b30112 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 ecefe7373..9d30fef35 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