Thirdparty
/
wvp-GB28181-pro
mirror of https://github.com/648540858/wvp-GB28181-pro
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'wvp-28181-2.0' into main-dev 

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
结构优化
648540858 2023-08-07 17:28:02 +08:00
parent 7079c73b99 0aada74b70
commit e898c344aa
27 changed files with 367 additions and 462 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/_content/introduction/config.md
View File

@ -160,4 +160,4 @@ user-settings:
如果配置信息无误，你可以启动zlm，再启动wvp来测试了，启动成功的话，你可以在wvp的日志下看到zlm已连接的提示。 如果配置信息无误，你可以启动zlm，再启动wvp来测试了，启动成功的话，你可以在wvp的日志下看到zlm已连接的提示。
接下来[部署到服务器](./_content/introduction/deployment.md)， 如何你只是本地运行直接再本地运行即可。 接下来[部署到服务器](./_content/introduction/deployment.md)， 如果你只是本地运行直接在本地运行即可。

20
src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java
View File

@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.conf; package com.genersoft.iot.vmp.conf;
import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
@ -45,6 +46,9 @@ public class DynamicTask {
* @return * @return
*/ */
public void startCron(String key, Runnable task, int cycleForCatalog) { public void startCron(String key, Runnable task, int cycleForCatalog) {
if(ObjectUtils.isEmpty(key)) {
return;
}
ScheduledFuture<?> future = futureMap.get(key); ScheduledFuture<?> future = futureMap.get(key);
if (future != null) { if (future != null) {
if (future.isCancelled()) { if (future.isCancelled()) {
@ -73,6 +77,9 @@ public class DynamicTask {
* @return * @return
*/ */
public void startDelay(String key, Runnable task, int delay) { public void startDelay(String key, Runnable task, int delay) {
if(ObjectUtils.isEmpty(key)) {
return;
}
stop(key); stop(key);
// 获取执行的时刻 // 获取执行的时刻
@ -99,9 +106,12 @@ public class DynamicTask {
} }
public boolean stop(String key) { public boolean stop(String key) {
if(ObjectUtils.isEmpty(key)) {
return false;
}
boolean result = false; boolean result = false;
if (futureMap.get(key) != null && !futureMap.get(key).isCancelled() && !futureMap.get(key).isDone()) { if (!ObjectUtils.isEmpty(futureMap.get(key)) && !futureMap.get(key).isCancelled() && !futureMap.get(key).isDone()) {
result = futureMap.get(key).cancel(false); result = futureMap.get(key).cancel(true);
futureMap.remove(key); futureMap.remove(key);
runnableMap.remove(key); runnableMap.remove(key);
} }
@ -109,6 +119,9 @@ public class DynamicTask {
} }
public boolean contains(String key) { public boolean contains(String key) {
if(ObjectUtils.isEmpty(key)) {
return false;
}
return futureMap.get(key) != null; return futureMap.get(key) != null;
} }
@ -117,6 +130,9 @@ public class DynamicTask {
} }
public Runnable get(String key) { public Runnable get(String key) {
if(ObjectUtils.isEmpty(key)) {
return null;
}
return runnableMap.get(key); return runnableMap.get(key);
} }

10
src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
View File

@ -43,8 +43,6 @@ public class UserSetting {
private Boolean pushAuthority = Boolean.TRUE; private Boolean pushAuthority = Boolean.TRUE;
private Boolean gbSendStreamStrict = Boolean.FALSE;
private Boolean syncChannelOnDeviceOnline = Boolean.FALSE; private Boolean syncChannelOnDeviceOnline = Boolean.FALSE;
private Boolean sipLog = Boolean.FALSE; private Boolean sipLog = Boolean.FALSE;
@ -208,14 +206,6 @@ public class UserSetting {
this.pushAuthority = pushAuthority; this.pushAuthority = pushAuthority;
} }
public Boolean getGbSendStreamStrict() {
return gbSendStreamStrict;
}
public void setGbSendStreamStrict(Boolean gbSendStreamStrict) {
this.gbSendStreamStrict = gbSendStreamStrict;
}
public Boolean getSyncChannelOnDeviceOnline() { public Boolean getSyncChannelOnDeviceOnline() {
return syncChannelOnDeviceOnline; return syncChannelOnDeviceOnline;
} }

2
src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java
View File

@ -165,7 +165,7 @@ public class Device {
* 是否开启ssrc校验，默认关闭，开启可以防止串流 * 是否开启ssrc校验，默认关闭，开启可以防止串流
*/ */
@Schema(description = "是否开启ssrc校验，默认关闭，开启可以防止串流") @Schema(description = "是否开启ssrc校验，默认关闭，开启可以防止串流")
private boolean ssrcCheck = true; private boolean ssrcCheck = false;
/** /**
* 地理坐标系， 目前支持 WGS84,GCJ02 * 地理坐标系， 目前支持 WGS84,GCJ02

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
View File

@ -333,7 +333,7 @@ public interface ISIPCommander {
* @param endTime 报警发生终止时间（可选） * @param endTime 报警发生终止时间（可选）
* @return true = 命令发送成功 * @return true = 命令发送成功
*/ */
void alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String alarmType, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException; void alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException;
/** /**
* 订阅、取消订阅目录信息 * 订阅、取消订阅目录信息

1
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
View File

@ -167,6 +167,7 @@ public class SIPRequestHeaderPlarformProvider {
public Request createMessageRequest(ParentPlatform parentPlatform, String content, SendRtpItem sendRtpItem) throws PeerUnavailableException, ParseException, InvalidArgumentException { public Request createMessageRequest(ParentPlatform parentPlatform, String content, SendRtpItem sendRtpItem) throws PeerUnavailableException, ParseException, InvalidArgumentException {
CallIdHeader callIdHeader = SipFactory.getInstance().createHeaderFactory().createCallIdHeader(sendRtpItem.getCallId()); CallIdHeader callIdHeader = SipFactory.getInstance().createHeaderFactory().createCallIdHeader(sendRtpItem.getCallId());
callIdHeader.setCallId(sendRtpItem.getCallId());
return createMessageRequest(parentPlatform, content, sendRtpItem.getToTag(), SipUtils.getNewViaTag(), sendRtpItem.getFromTag(), callIdHeader); return createMessageRequest(parentPlatform, content, sendRtpItem.getToTag(), SipUtils.getNewViaTag(), sendRtpItem.getFromTag(), callIdHeader);
} }

5
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@ -1275,7 +1275,7 @@ public class SIPCommander implements ISIPCommander {
* @return true = 命令发送成功 * @return true = 命令发送成功
*/ */
@Override @Override
public void alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String alarmType, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException { public void alarmSubscribe(Device device, int expires, String startPriority, String endPriority, String alarmMethod, String startTime, String endTime) throws InvalidArgumentException, SipException, ParseException {
StringBuffer cmdXml = new StringBuffer(200); StringBuffer cmdXml = new StringBuffer(200);
String charset = device.getCharset(); String charset = device.getCharset();
@ -1293,9 +1293,6 @@ public class SIPCommander implements ISIPCommander {
if (!ObjectUtils.isEmpty(alarmMethod)) { if (!ObjectUtils.isEmpty(alarmMethod)) {
cmdXml.append("<AlarmMethod>" + alarmMethod + "</AlarmMethod>\r\n"); cmdXml.append("<AlarmMethod>" + alarmMethod + "</AlarmMethod>\r\n");
} }
if (!ObjectUtils.isEmpty(alarmType)) {
cmdXml.append("<AlarmType>" + alarmType + "</AlarmType>\r\n");
}
if (!ObjectUtils.isEmpty(startTime)) { if (!ObjectUtils.isEmpty(startTime)) {
cmdXml.append("<StartAlarmTime>" + startTime + "</StartAlarmTime>\r\n"); cmdXml.append("<StartAlarmTime>" + startTime + "</StartAlarmTime>\r\n");
} }

21
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
View File

@ -98,15 +98,20 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
logger.warn("[收到ACK]：未找到来自{}，目标为({})的推流信息",fromUserId, toUserId); logger.warn("[收到ACK]：未找到来自{}，目标为({})的推流信息",fromUserId, toUserId);
return; return;
} }
// tcp主动时，此时是级联下级平台，在回复200ok时，本地已经请求zlm开启监听，跳过下面步骤 // tcp主动时，此时是级联下级平台，在回复200ok时，本地已经请求zlm开启监听，跳过下面步骤
if (sendRtpItem.isTcpActive()) { if (sendRtpItem.isTcpActive()) {
return; logger.info("收到ACK，rtp/{} TCP主动方式后续处理", sendRtpItem.getStreamId());
} return;
logger.info("[收到ACK]：rtp/{}开始级推流, 目标={}:{}，SSRC={}, RTCP={}", sendRtpItem.getStream(), }
sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp()); String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
// 取消设置的超时任务
dynamicTask.stop(callIdHeader.getCallId());
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
logger.info("收到ACK，rtp/{}开始向上级推流, 目标={}:{}，SSRC={}, 协议:{}",
sendRtpItem.getStream(),
sendRtpItem.getIp(),
sendRtpItem.getPort(),
sendRtpItem.getSsrc(),
sendRtpItem.isTcp()?(sendRtpItem.isTcpActive()?"TCP主动":"TCP被动"):"UDP"
);
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(fromUserId); ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(fromUserId);
if (parentPlatform != null) { if (parentPlatform != null) {

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
View File

@ -148,7 +148,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId); logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
} }
try { try {
logger.warn("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); logger.info("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null); cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null);
} catch (InvalidArgumentException | ParseException | SipException | } catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) { SsrcTransactionNotFoundException e) {

13
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
View File

@ -469,8 +469,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
sendRtpItem.setApp("rtp"); sendRtpItem.setApp("rtp");
if ("Playback".equalsIgnoreCase(sessionName)) { if ("Playback".equalsIgnoreCase(sessionName)) {
sendRtpItem.setPlayType(InviteStreamType.PLAYBACK); sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam()); String startTimeStr = DateUtil.urlFormatter.format(start);
sendRtpItem.setStream(ssrcInfo.getStream()); 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());
// 写入redis， 超时时回复 // 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem); redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start), playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
@ -530,12 +532,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
} }
})); }));
sendRtpItem.setPlayType(InviteStreamType.PLAY); sendRtpItem.setPlayType(InviteStreamType.PLAY);
String streamId = null; String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}else {
streamId = String.format("%08x", Integer.parseInt(ssrcInfo.getSsrc())).toUpperCase();
}
sendRtpItem.setStream(streamId); sendRtpItem.setStream(streamId);
sendRtpItem.setSsrc(ssrcInfo.getSsrc()); sendRtpItem.setSsrc(ssrcInfo.getSsrc());
redisCatchStorage.updateSendRTPSever(sendRtpItem); redisCatchStorage.updateSendRTPSever(sendRtpItem);

11
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java
View File

@ -33,7 +33,7 @@ import java.text.ParseException;
public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler { public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler {
private Logger logger = LoggerFactory.getLogger(KeepaliveNotifyMessageHandler.class); private final Logger logger = LoggerFactory.getLogger(KeepaliveNotifyMessageHandler.class);
private final static String cmdType = "Keepalive"; private final static String cmdType = "Keepalive";
@Autowired @Autowired
@ -59,14 +59,19 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
// 未注册的设备不做处理 // 未注册的设备不做处理
return; return;
} }
logger.info("[收到心跳]， device: {}", device.getDeviceId());
SIPRequest request = (SIPRequest) evt.getRequest(); SIPRequest request = (SIPRequest) evt.getRequest();
logger.info("[收到心跳]， device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
// 回复200 OK // 回复200 OK
try { try {
responseAck(request, Response.OK); responseAck(request, Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 心跳回复: {}", e.getMessage()); logger.error("[命令发送失败] 心跳回复: {}", e.getMessage());
} }
if (DateUtil.getDifferenceForNow(device.getKeepaliveTime()) <= 3000L){
logger.info("[收到心跳] 心跳发送过于频繁，已忽略 device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
return;
}
RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress()); RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress());
if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) { if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) {
@ -80,7 +85,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
}else { }else {
long lastTime = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(device.getKeepaliveTime()); long lastTime = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(device.getKeepaliveTime());
if (System.currentTimeMillis()/1000-lastTime > 10) { if (System.currentTimeMillis()/1000-lastTime > 10) {
device.setKeepaliveIntervalTime(new Long(System.currentTimeMillis()/1000-lastTime).intValue()); device.setKeepaliveIntervalTime(Long.valueOf(System.currentTimeMillis()/1000-lastTime).intValue());
} }
} }

2
src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java
View File

@ -281,6 +281,6 @@ public class SipUtils {
return null; return null;
} }
} }
return localDateTime.format(DateUtil.formatterISO8601); return localDateTime.format(DateUtil.formatter);
} }
} }

19
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@ -200,7 +200,10 @@ public class ZLMHttpHookListener {
String mediaServerId = json.getString("mediaServerId"); String mediaServerId = json.getString("mediaServerId");
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
if (mediaInfo == null) {
return new HookResultForOnPublish(200, "success");
}
// 推流鉴权的处理
if (!"rtp".equals(param.getApp())) { if (!"rtp".equals(param.getApp())) {
if (userSetting.getPushAuthority()) { if (userSetting.getPushAuthority()) {
// 推流鉴权 // 推流鉴权
@ -252,11 +255,21 @@ public class ZLMHttpHookListener {
} }
}); });
// 是否录像
if ("rtp".equals(param.getApp())) { if ("rtp".equals(param.getApp())) {
result.setEnable_mp4(userSetting.getRecordSip()); result.setEnable_mp4(userSetting.getRecordSip());
} else { } else {
result.setEnable_mp4(userSetting.isRecordPushLive()); 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<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream()); List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) { if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
String deviceId = ssrcTransactionForAll.get(0).getDeviceId(); String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
@ -569,6 +582,7 @@ public class ZLMHttpHookListener {
Device device = deviceService.getDevice(inviteInfo.getDeviceId()); Device device = deviceService.getDevice(inviteInfo.getDeviceId());
if (device != null) { if (device != null) {
try { try {
// 多查询一次防止已经被处理了
InviteInfo info = inviteStreamService.getInviteInfo(inviteInfo.getType(), InviteInfo info = inviteStreamService.getInviteInfo(inviteInfo.getType(),
inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());
if (info != null) { if (info != null) {
@ -643,7 +657,7 @@ public class ZLMHttpHookListener {
if ("rtp".equals(param.getApp())) { if ("rtp".equals(param.getApp())) {
String[] s = param.getStream().split("_"); 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()); defaultResult.setResult(HookResult.SUCCESS());
return defaultResult; return defaultResult;
} }
@ -672,7 +686,6 @@ public class ZLMHttpHookListener {
result.onTimeout(() -> { result.onTimeout(() -> {
logger.info("[ZLM HOOK] 预览流自动点播, 等待超时"); logger.info("[ZLM HOOK] 预览流自动点播, 等待超时");
// 释放rtpserver
msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时")); msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));
resultHolder.invokeResult(msg); resultHolder.invokeResult(msg);
}); });

21
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java
View File

@ -168,13 +168,9 @@ public class ZLMServerFactory {
public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId,
String deviceId, String channelId, boolean tcp, boolean rtcp){ String deviceId, String channelId, boolean tcp, boolean rtcp){
// 默认为随机端口 int localPort = sendRtpPortManager.getNextPort(serverItem);
int localPort = 0; if (localPort == 0) {
if (userSetting.getGbSendStreamStrict()) { return null;
localPort = sendRtpPortManager.getNextPort(serverItem);
if (localPort == 0) {
return null;
}
} }
SendRtpItem sendRtpItem = new SendRtpItem(); SendRtpItem sendRtpItem = new SendRtpItem();
sendRtpItem.setIp(ip); sendRtpItem.setIp(ip);
@ -204,13 +200,10 @@ public class ZLMServerFactory {
*/ */
public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId,
String app, String stream, String channelId, boolean tcp, boolean rtcp){ String app, String stream, String channelId, boolean tcp, boolean rtcp){
// 默认为随机端口
int localPort = 0; int localPort = sendRtpPortManager.getNextPort(serverItem);
if (userSetting.getGbSendStreamStrict()) { if (localPort == 0) {
localPort = sendRtpPortManager.getNextPort(serverItem); return null;
if (localPort == 0) {
return null;
}
} }
SendRtpItem sendRtpItem = new SendRtpItem(); SendRtpItem sendRtpItem = new SendRtpItem();
sendRtpItem.setIp(ip); sendRtpItem.setIp(ip);

10
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java
View File

@ -6,6 +6,7 @@ public class HookResultForOnPublish extends HookResult{
private boolean enable_mp4; private boolean enable_mp4;
private int mp4_max_second; private int mp4_max_second;
private String mp4_save_path; private String mp4_save_path;
private String stream_replace;
public HookResultForOnPublish() { public HookResultForOnPublish() {
} }
@ -51,12 +52,21 @@ public class HookResultForOnPublish extends HookResult{
this.mp4_save_path = mp4_save_path; 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 @Override
public String toString() { public String toString() {
return "HookResultForOnPublish{" + return "HookResultForOnPublish{" +
"enable_audio=" + enable_audio + "enable_audio=" + enable_audio +
", enable_mp4=" + enable_mp4 + ", enable_mp4=" + enable_mp4 +
", mp4_max_second=" + mp4_max_second + ", mp4_max_second=" + mp4_max_second +
", stream_replace=" + stream_replace +
", mp4_save_path='" + mp4_save_path + '\'' + ", mp4_save_path='" + mp4_save_path + '\'' +
'}'; '}';
} }

8
src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java
View File

@ -74,5 +74,13 @@ public interface IInviteStreamService {
int getStreamInfoCount(String mediaServerId); int getStreamInfoCount(String mediaServerId);
/**
* 获取MediaServer下的流信息
*/
InviteInfo getInviteInfoBySSRC(String ssrc);
/**
* 更新ssrc
*/
InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrcInResponse);
} }

11
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
View File

@ -190,6 +190,17 @@ public class DeviceServiceImpl implements IDeviceService {
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true); redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
} }
//
// try {
// cmder.alarmSubscribe(device, 600, "0", "4", "0", "2023-7-27T00:00:00", "2023-7-28T00:00:00");
// } catch (InvalidArgumentException e) {
// throw new RuntimeException(e);
// } catch (SipException e) {
// throw new RuntimeException(e);
// } catch (ParseException e) {
// throw new RuntimeException(e);
// }
} }
@Override @Override

80
src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java
View File

@ -77,10 +77,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
} }
String key = VideoManagerConstants.INVITE_PREFIX + String key = VideoManagerConstants.INVITE_PREFIX +
"_" + inviteInfoForUpdate.getType() + ":" + inviteInfoForUpdate.getType() +
"_" + inviteInfoForUpdate.getDeviceId() + ":" + inviteInfoForUpdate.getDeviceId() +
"_" + inviteInfoForUpdate.getChannelId() + ":" + inviteInfoForUpdate.getChannelId() +
"_" + inviteInfoForUpdate.getStream(); ":" + inviteInfoForUpdate.getStream()+
":" + inviteInfoForUpdate.getSsrcInfo().getSsrc();
redisTemplate.opsForValue().set(key, inviteInfoForUpdate); redisTemplate.opsForValue().set(key, inviteInfoForUpdate);
} }
@ -93,11 +94,15 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
} }
removeInviteInfo(inviteInfoInDb); removeInviteInfo(inviteInfoInDb);
String key = VideoManagerConstants.INVITE_PREFIX + String key = VideoManagerConstants.INVITE_PREFIX +
"_" + inviteInfo.getType() + ":" + inviteInfo.getType() +
"_" + inviteInfo.getDeviceId() + ":" + inviteInfo.getDeviceId() +
"_" + inviteInfo.getChannelId() + ":" + inviteInfo.getChannelId() +
"_" + stream; ":" + stream +
":" + inviteInfo.getSsrcInfo().getSsrc();
inviteInfoInDb.setStream(stream); inviteInfoInDb.setStream(stream);
if (inviteInfoInDb.getSsrcInfo() != null) {
inviteInfoInDb.getSsrcInfo().setStream(stream);
}
redisTemplate.opsForValue().set(key, inviteInfoInDb); redisTemplate.opsForValue().set(key, inviteInfoInDb);
return inviteInfoInDb; return inviteInfoInDb;
} }
@ -105,10 +110,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Override @Override
public InviteInfo getInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) { public InviteInfo getInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = VideoManagerConstants.INVITE_PREFIX + String key = VideoManagerConstants.INVITE_PREFIX +
"_" + (type != null ? type : "*") + ":" + (type != null ? type : "*") +
"_" + (deviceId != null ? deviceId : "*") + ":" + (deviceId != null ? deviceId : "*") +
"_" + (channelId != null ? channelId : "*") + ":" + (channelId != null ? channelId : "*") +
"_" + (stream != null ? stream : "*"); ":" + (stream != null ? stream : "*")
+ ":*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key); List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() != 1) { if (scanResult.size() != 1) {
return null; return null;
@ -130,10 +136,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Override @Override
public void removeInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) { public void removeInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) {
String scanKey = VideoManagerConstants.INVITE_PREFIX + String scanKey = VideoManagerConstants.INVITE_PREFIX +
"_" + (type != null ? type : "*") + ":" + (type != null ? type : "*") +
"_" + (deviceId != null ? deviceId : "*") + ":" + (deviceId != null ? deviceId : "*") +
"_" + (channelId != null ? channelId : "*") + ":" + (channelId != null ? channelId : "*") +
"_" + (stream != null ? stream : "*"); ":" + (stream != null ? stream : "*") +
":*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, scanKey); List<Object> scanResult = RedisUtil.scan(redisTemplate, scanKey);
if (scanResult.size() > 0) { if (scanResult.size() > 0) {
for (Object keyObj : scanResult) { for (Object keyObj : scanResult) {
@ -171,10 +178,10 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
} }
private String buildKey(InviteSessionType type, String deviceId, String channelId, String stream) { private String buildKey(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = type + "_" + deviceId + "_" + channelId; String key = type + ":" + deviceId + ":" + channelId;
// 如果ssrc未null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite // 如果ssrc未null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite
if (stream != null) { if (stream != null) {
key += ("_" + stream); key += (":" + stream);
} }
return key; return key;
} }
@ -188,7 +195,7 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
@Override @Override
public int getStreamInfoCount(String mediaServerId) { public int getStreamInfoCount(String mediaServerId) {
int count = 0; int count = 0;
String key = VideoManagerConstants.INVITE_PREFIX + "_*_*_*_*"; String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key); List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() == 0) { if (scanResult.size() == 0) {
return 0; return 0;
@ -219,11 +226,42 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
private String buildSubStreamKey(InviteSessionType type, String deviceId, String channelId, String stream) { private String buildSubStreamKey(InviteSessionType type, String deviceId, String channelId, String stream) {
String key = type + "_" + "_" + deviceId + "_" + channelId; String key = type + ":" + ":" + deviceId + ":" + channelId;
// 如果ssrc为null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite // 如果ssrc为null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite
if (stream != null) { if (stream != null) {
key += ("_" + stream); key += (":" + stream);
} }
return key; return key;
} }
@Override
public InviteInfo getInviteInfoBySSRC(String ssrc) {
String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:" + ssrc;
List<Object> 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;
}
} }

7
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
View File

@ -153,9 +153,14 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (streamId == null) { if (streamId == null) {
streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase(); streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
} }
int ssrcCheckParam = 0;
if (ssrcCheck && tcpMode > 1) {
// 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验
logger.warn("[openRTPServer] TCP被动/TCP主动收流时，默认关闭ssrc检验");
}
int rtpServerPort; int rtpServerPort;
if (mediaServerItem.isRtpEnable()) { if (mediaServerItem.isRtpEnable()) {
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port,onlyAuto, reUsePort, tcpMode); rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0)?Integer.parseInt(ssrc):0, port, onlyAuto, reUsePort, tcpMode);
} else { } else {
rtpServerPort = mediaServerItem.getRtpProxyPort(); rtpServerPort = mediaServerItem.getRtpProxyPort();
} }

524
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@ -18,6 +18,8 @@ import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.zlm.*; import com.genersoft.iot.vmp.media.zlm.*;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
@ -43,6 +45,7 @@ import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent; import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent;
import gov.nist.javax.sip.message.SIPResponse; import gov.nist.javax.sip.message.SIPResponse;
import gov.nist.javax.sip.message.SIPResponse;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@ -147,15 +150,21 @@ public class PlayServiceImpl implements IPlayService {
@Override @Override
public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, String ssrc, ErrorCallback<Object> callback) { public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, String ssrc, ErrorCallback<Object> callback) {
if (mediaServerItem == null) { if (mediaServerItem == null) {
logger.warn("[点播] 未找到可用的zlm deviceId: {},channelId:{}", deviceId, channelId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm");
} }
Device device = redisCatchStorage.getDevice(deviceId); Device device = redisCatchStorage.getDevice(deviceId);
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE") && !mediaServerItem.isRtpEnable()) {
logger.warn("[点播] 单端口收流时不支持TCP主动方式收流 deviceId: {},channelId:{}", deviceId, channelId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "单端口收流时不支持TCP主动方式收流");
}
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
if (inviteInfo != null ) { if (inviteInfo != null ) {
if (inviteInfo.getStreamInfo() == null) { if (inviteInfo.getStreamInfo() == null) {
// 点播发起了但是尚未成功, 仅注册回调等待结果即可 // 点播发起了但是尚未成功, 仅注册回调等待结果即可
inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback);
logger.info("[点播开始] 已经请求中，等待结果， deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
return inviteInfo.getSsrcInfo(); return inviteInfo.getSsrcInfo();
}else { }else {
StreamInfo streamInfo = inviteInfo.getStreamInfo(); StreamInfo streamInfo = inviteInfo.getStreamInfo();
@ -178,6 +187,7 @@ public class PlayServiceImpl implements IPlayService {
InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getCode(),
InviteErrorCode.SUCCESS.getMsg(), InviteErrorCode.SUCCESS.getMsg(),
streamInfo); streamInfo);
logger.info("[点播已存在] 直接返回， deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
return inviteInfo.getSsrcInfo(); return inviteInfo.getSsrcInfo();
}else { }else {
// 点播发起了但是尚未成功, 仅注册回调等待结果即可 // 点播发起了但是尚未成功, 仅注册回调等待结果即可
@ -187,11 +197,8 @@ public class PlayServiceImpl implements IPlayService {
} }
} }
} }
String streamId = null; String streamId = String.format("%s_%s", device.getDeviceId(), channelId);;
if (mediaServerItem.isRtpEnable()) { SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam());
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, false,device.getStreamModeForParam());
if (ssrcInfo == null) { if (ssrcInfo == null) {
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null); callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
@ -200,7 +207,6 @@ public class PlayServiceImpl implements IPlayService {
null); null);
return null; return null;
} }
// TODO 记录点播的状态
play(mediaServerItem, ssrcInfo, device, channelId, callback); play(mediaServerItem, ssrcInfo, device, channelId, callback);
return ssrcInfo; return ssrcInfo;
} }
@ -357,8 +363,8 @@ public class PlayServiceImpl implements IPlayService {
null); null);
return; return;
} }
logger.info("[点播开始] deviceId: {}, channelId: {},码流类型：{}, 收流端口： {}, 收流模式：{}, SSRC: {}, SSRC校验：{}", logger.info("[点播开始] deviceId: {}, channelId: {},码流类型：{}, 收流端口： {}, STREAM：{}, 收流模式：{}, SSRC: {}, SSRC校验：{}",
device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流", ssrcInfo.getPort(), device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流", ssrcInfo.getPort(), ssrcInfo.getStream(),
device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
//端口获取失败的ssrcInfo 没有必要发送点播指令 //端口获取失败的ssrcInfo 没有必要发送点播指令
if (ssrcInfo.getPort() <= 0) { if (ssrcInfo.getPort() <= 0) {
@ -389,16 +395,6 @@ public class PlayServiceImpl implements IPlayService {
device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流", device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流",
ssrcInfo.getPort(), ssrcInfo.getSsrc()); ssrcInfo.getPort(), ssrcInfo.getSsrc());
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
// InviteInfo inviteInfoForTimeout = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.play, device.getDeviceId(), channelId);
// if (inviteInfoForTimeout == null) {
// return;
// }
// if (InviteSessionStatus.ok == inviteInfoForTimeout.getStatus() ) {
// // TODO 发送bye
// }else {
// // TODO 发送cancel
// }
callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
@ -443,128 +439,10 @@ public class PlayServiceImpl implements IPlayService {
logger.info("[点播成功] deviceId: {}, channelId:{}, 码流类型：{}", device.getDeviceId(), channelId, logger.info("[点播成功] deviceId: {}, channelId:{}, 码流类型：{}", device.getDeviceId(), channelId,
device.isSwitchPrimarySubStream() ? "辅码流" : "主码流"); device.isSwitchPrimarySubStream() ? "辅码流" : "主码流");
snapOnPlay(mediaServerItemInuse, device.getDeviceId(), channelId, ssrcInfo.getStream()); snapOnPlay(mediaServerItemInuse, device.getDeviceId(), channelId, ssrcInfo.getStream());
}, (event) -> { }, (eventResult) -> {
inviteInfo.setStatus(InviteSessionStatus.ok); // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
ResponseEvent responseEvent = (ResponseEvent) event.event; timeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAY);
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).trim();
// 查询到ssrc不一致且开启了ssrc校验则需要针对处理
if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
String substring = contentString.substring(0, contentString.indexOf("y="));
try {
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
int port = -1;
Vector mediaDescriptions = sdp.getMediaDescriptions(true);
for (Object description : mediaDescriptions) {
MediaDescription mediaDescription = (MediaDescription) description;
Media media = mediaDescription.getMedia();
Vector mediaFormats = media.getMediaFormats(false);
if (mediaFormats.contains("96")) {
port = media.getMediaPort();
break;
}
}
logger.info("[点播-TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址：{}:{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
JSONObject jsonObject = zlmresTfulUtils.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream());
logger.info("[点播-TCP主动连接对方] 结果： {}", jsonObject);
} catch (SdpException e) {
logger.error("[点播-TCP主动连接对方] deviceId: {}, channelId: {}, 解析200OK的SDP信息失败", device.getDeviceId(), channelId, e);
dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
}
}
return;
}
logger.info("[点播消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
logger.info("[点播消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
// 单端口模式streamId也有变化，重新设置监听即可
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase();
hookSubscribe.getContent().put("stream", stream);
inviteStreamService.updateInviteInfoForStream(inviteInfo, stream);
subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> {
logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + hookParam);
dynamicTask.stop(timeOutTaskKey);
// 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.PLAY, 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.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.SUCCESS.getCode(),
InviteErrorCode.SUCCESS.getMsg(),
streamInfo);
snapOnPlay(mediaServerItemInUse, device.getDeviceId(), channelId, stream);
});
return;
}
// 更新ssrc
Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse);
if (!result) {
try {
logger.warn("[点播] 更新ssrc失败，停止点播 {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
}
dynamicTask.stop(timeOutTaskKey);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
"下级自定义了ssrc,重新设置收流信息失败", null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
"下级自定义了ssrc,重新设置收流信息失败", null);
}else {
if (ssrcInfo.getStream()!= null && !ssrcInfo.getStream().equals(inviteInfo.getStream())) {
inviteStreamService.removeInviteInfo(inviteInfo);
}
ssrcInfo.setSsrc(ssrcInResponse);
inviteInfo.setSsrcInfo(ssrcInfo);
inviteInfo.setStream(ssrcInfo.getStream());
}
}else {
logger.info("[点播消息] 收到invite 200, 下级自定义了ssrc, 但是当前模式无需修正");
}
}
inviteStreamService.updateInviteInfo(inviteInfo);
}, (event) -> { }, (event) -> {
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
@ -601,6 +479,47 @@ public class PlayServiceImpl implements IPlayService {
} }
} }
private void tcpActiveHandler(Device device, String channelId, String contentString,
MediaServerItem mediaServerItem,
String timeOutTaskKey, SSRCInfo ssrcInfo, ErrorCallback<Object> callback){
if (!device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
return;
}
String substring = contentString.substring(0, contentString.indexOf("y="));
try {
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
int port = -1;
Vector mediaDescriptions = sdp.getMediaDescriptions(true);
for (Object description : mediaDescriptions) {
MediaDescription mediaDescription = (MediaDescription) description;
Media media = mediaDescription.getMedia();
Vector mediaFormats = media.getMediaFormats(false);
if (mediaFormats.contains("96")) {
port = media.getMediaPort();
break;
}
}
logger.info("[TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址：{}:{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
JSONObject jsonObject = zlmresTfulUtils.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream());
logger.info("[TCP主动连接对方] 结果： {}", jsonObject);
} catch (SdpException e) {
logger.error("[TCP主动连接对方] deviceId: {}, channelId: {}, 解析200OK的SDP信息失败", device.getDeviceId(), channelId, e);
dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
}
}
/** /**
* 点播成功时调用截图. * 点播成功时调用截图.
* *
@ -707,22 +626,23 @@ public class PlayServiceImpl implements IPlayService {
String endTime, ErrorCallback<Object> callback) { String endTime, ErrorCallback<Object> callback) {
Device device = storager.queryVideoDevice(deviceId); Device device = storager.queryVideoDevice(deviceId);
if (device == null) { if (device == null) {
return; logger.warn("[录像回放] 未找到设备 deviceId: {},channelId:{}", deviceId, channelId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到设备：" + deviceId);
} }
MediaServerItem newMediaServerItem = getNewMediaServerItem(device); MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
String stream = null; if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE") && ! newMediaServerItem.isRtpEnable()) {
if (newMediaServerItem.isRtpEnable()) { logger.warn("[录像回放] 单端口收流时不支持TCP主动方式收流 deviceId: {},channelId:{}", deviceId, channelId);
String startTimeStr = startTime.replace("-", "") throw new ControllerException(ErrorCode.ERROR100.getCode(), "单端口收流时不支持TCP主动方式收流");
.replace(":", "")
.replace(" ", "");
System.out.println(startTimeStr);
String endTimeTimeStr = endTime.replace("-", "")
.replace(":", "")
.replace(" ", "");
System.out.println(endTimeTimeStr);
stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
} }
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false,false, device.getStreamModeForParam()); 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); playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, callback);
} }
@ -795,113 +715,13 @@ public class PlayServiceImpl implements IPlayService {
try { try {
cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime,
hookEvent, eventResult -> { hookEvent, eventResult -> {
inviteInfo.setStatus(InviteSessionStatus.ok); // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
ResponseEvent responseEvent = (ResponseEvent) eventResult.event; InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
String contentString = new String(responseEvent.getResponse().getRawContent()); playBackTimeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAYBACK);
// 获取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)) {
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
String substring = contentString.substring(0, contentString.indexOf("y="));
try {
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
int port = -1;
Vector mediaDescriptions = sdp.getMediaDescriptions(true);
for (Object description : mediaDescriptions) {
MediaDescription mediaDescription = (MediaDescription) description;
Media media = mediaDescription.getMedia();
Vector mediaFormats = media.getMediaFormats(false);
if (mediaFormats.contains("96")) {
port = media.getMediaPort();
break;
}
}
logger.info("[录像回放-TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址：{}:{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
JSONObject jsonObject = zlmresTfulUtils.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream());
logger.info("[录像回放-TCP主动连接对方] 结果： {}", jsonObject);
} catch (SdpException e) {
logger.error("[录像回放-TCP主动连接对方] deviceId: {}, channelId: {}, 解析200OK的SDP信息失败", device.getDeviceId(), channelId, e);
dynamicTask.stop(playBackTimeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
}
}
return;
}
logger.info("[录像回放] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
logger.info("[录像回放] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
// 单端口模式streamId也有变化，需要重新设置监听
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase();
hookSubscribe.getContent().put("stream", stream);
inviteStreamService.updateInviteInfoForStream(inviteInfo, stream);
subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> {
logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + hookParam);
dynamicTask.stop(playBackTimeOutTaskKey);
// hook响应
hookEvent.response(mediaServerItemInUse, hookParam);
});
}
// 更新ssrc
Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse);
if (!result) {
try {
logger.warn("[录像回放] 更新ssrc失败，停止录像回放 {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
}
dynamicTask.stop(playBackTimeOutTaskKey);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
"下级自定义了ssrc,重新设置收流信息失败", null);
}else {
if (ssrcInfo.getStream()!= null && !ssrcInfo.getStream().equals(inviteInfo.getStream())) {
inviteStreamService.removeInviteInfo(inviteInfo);
}
ssrcInfo.setSsrc(ssrcInResponse);
inviteInfo.setSsrcInfo(ssrcInfo);
inviteInfo.setStream(ssrcInfo.getStream());
}
}else {
logger.info("[点播消息] 收到invite 200, 下级自定义了ssrc, 但是当前模式无需修正");
}
}
inviteStreamService.updateInviteInfo(inviteInfo);
}, errorEvent); }, errorEvent);
} catch (InvalidArgumentException | SipException | ParseException e) { } catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 回放: {}", e.getMessage()); logger.error("[命令发送失败] 录像回放: {}", e.getMessage());
SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent; eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent;
@ -912,6 +732,90 @@ public class PlayServiceImpl implements IPlayService {
} }
private void InviteOKHandler(SipSubscribe.EventResult eventResult, SSRCInfo ssrcInfo, MediaServerItem mediaServerItem,
Device device, String channelId, String timeOutTaskKey, ErrorCallback<Object> callback,
InviteInfo inviteInfo, InviteSessionType inviteSessionType){
inviteInfo.setStatus(InviteSessionStatus.ok);
ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
String contentString = new String(responseEvent.getResponse().getRawContent());
String ssrcInResponse = SipUtils.getSsrcFromSdp(contentString);
if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
// ssrc 一致
if (mediaServerItem.isRtpEnable()) {
// 多端口
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
tcpActiveHandler(device, channelId, contentString, mediaServerItem, timeOutTaskKey, ssrcInfo, callback);
}
}else {
// 单端口
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
logger.warn("[Invite 200OK] 单端口收流模式不支持tcp主动模式收流");
}
}
}else {
logger.info("[Invite 200OK] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
// ssrc 不一致
if (mediaServerItem.isRtpEnable()) {
// 多端口
if (device.isSsrcCheck()) {
// ssrc检验
// 更新ssrc
logger.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse);
if (!result) {
try {
logger.warn("[Invite 200OK] 更新ssrc失败，停止点播 {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止播放， 发送BYE: {}", e.getMessage());
}
dynamicTask.stop(timeOutTaskKey);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
"下级自定义了ssrc,重新设置收流信息失败", null);
inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
"下级自定义了ssrc,重新设置收流信息失败", null);
}else {
ssrcInfo.setSsrc(ssrcInResponse);
inviteInfo.setSsrcInfo(ssrcInfo);
inviteInfo.setStream(ssrcInfo.getStream());
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
if (mediaServerItem.isRtpEnable()) {
tcpActiveHandler(device, channelId, contentString, mediaServerItem, timeOutTaskKey, ssrcInfo, callback);
}else {
logger.warn("[Invite 200OK] 单端口收流模式不支持tcp主动模式收流");
}
}
inviteStreamService.updateInviteInfo(inviteInfo);
}
}
}else {
if (ssrcInResponse != null) {
// 单端口
// 重新订阅流上线
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), null, inviteInfo.getStream());
streamSession.remove(inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), inviteInfo.getStream());
inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse);
streamSession.put(device.getDeviceId(), channelId, ssrcTransaction.getCallId(),
inviteInfo.getStream(), ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType);
}
}
}
}
@Override @Override
public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, ErrorCallback<Object> callback) { public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, ErrorCallback<Object> callback) {
Device device = storager.queryVideoDevice(deviceId); Device device = storager.queryVideoDevice(deviceId);
@ -925,6 +829,7 @@ public class PlayServiceImpl implements IPlayService {
null); null);
return; return;
} }
// 录像下载不使用固定流地址，固定流地址会导致如果开始时间与结束时间一致时文件错误的叠加在一起
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,false, device.getStreamModeForParam()); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,false, device.getStreamModeForParam());
download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, callback); download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, callback);
} }
@ -993,108 +898,9 @@ public class PlayServiceImpl implements IPlayService {
try { try {
cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed,
hookEvent, errorEvent, eventResult ->{ hookEvent, errorEvent, eventResult ->{
inviteInfo.setStatus(InviteSessionStatus.ok); // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
ResponseEvent responseEvent = (ResponseEvent) eventResult.event; InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
String contentString = new String(responseEvent.getResponse().getRawContent()); downLoadTimeOutTaskKey, callback, inviteInfo, InviteSessionType.DOWNLOAD);
// 获取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)) {
if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
String substring = contentString.substring(0, contentString.indexOf("y="));
try {
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
int port = -1;
Vector mediaDescriptions = sdp.getMediaDescriptions(true);
for (Object description : mediaDescriptions) {
MediaDescription mediaDescription = (MediaDescription) description;
Media media = mediaDescription.getMedia();
Vector mediaFormats = media.getMediaFormats(false);
if (mediaFormats.contains("96")) {
port = media.getMediaPort();
break;
}
}
logger.info("[录像下载-TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址：{}:{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
JSONObject jsonObject = zlmresTfulUtils.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream());
logger.info("[录像下载-TCP主动连接对方] 结果： {}", jsonObject);
} catch (SdpException e) {
logger.error("[录像下载-TCP主动连接对方] deviceId: {}, channelId: {}, 解析200OK的SDP信息失败", device.getDeviceId(), channelId, e);
dynamicTask.stop(downLoadTimeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null);
}
}
return;
}
logger.info("[录像下载] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
logger.info("[录像下载] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
// 单端口模式streamId也有变化，需要重新设置监听
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase();
hookSubscribe.getContent().put("stream", stream);
inviteStreamService.updateInviteInfoForStream(inviteInfo, stream);
subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> {
logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + hookParam);
dynamicTask.stop(downLoadTimeOutTaskKey);
hookEvent.response(mediaServerItemInUse, hookParam);
});
}
// 更新ssrc
Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse);
if (!result) {
try {
logger.warn("[录像下载] 更新ssrc失败，停止录像回放 {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
}
dynamicTask.stop(downLoadTimeOutTaskKey);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
"下级自定义了ssrc,重新设置收流信息失败", null);
}else {
if (ssrcInfo.getStream()!= null && !ssrcInfo.getStream().equals(inviteInfo.getStream())) {
inviteStreamService.removeInviteInfo(inviteInfo);
}
ssrcInfo.setSsrc(ssrcInResponse);
inviteInfo.setSsrcInfo(ssrcInfo);
inviteInfo.setStream(ssrcInfo.getStream());
}
}else {
logger.info("[录像下载] 收到invite 200, 下级自定义了ssrc, 但是当前模式无需修正");
}
}
inviteStreamService.updateInviteInfo(inviteInfo);
}); });
} catch (InvalidArgumentException | SipException | ParseException e) { } catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 录像下载: {}", e.getMessage()); logger.error("[命令发送失败] 录像下载: {}", e.getMessage());

6
src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java
View File

@ -7,6 +7,7 @@ import java.time.LocalDateTime;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException; import java.time.format.DateTimeParseException;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalAccessor; import java.time.temporal.TemporalAccessor;
import java.util.Locale; import java.util.Locale;
@ -106,4 +107,9 @@ public class DateUtil {
LocalDateTime nowDateTime = LocalDateTime.now(); LocalDateTime nowDateTime = LocalDateTime.now();
return formatterISO8601.format(nowDateTime); return formatterISO8601.format(nowDateTime);
} }
public static long getDifferenceForNow(String keepaliveTime) {
Instant beforeInstant = Instant.from(formatter.parse(keepaliveTime));
return ChronoUnit.MILLIS.between(beforeInstant, Instant.now());
}
} }

2
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java
View File

@ -67,7 +67,7 @@ public class MediaController {
&& streamAuthorityInfo.getCallId().equals(callId)) { && streamAuthorityInfo.getCallId().equals(callId)) {
authority = true; authority = true;
}else { }else {
throw new ControllerException(ErrorCode.ERROR400); throw new ControllerException(ErrorCode.ERROR400.getCode(), "获取播放地址鉴权失败");
} }
}else { }else {
// 是否登陆用户, 登陆用户返回完整信息 // 是否登陆用户, 登陆用户返回完整信息

11
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
View File

@ -97,6 +97,7 @@ public class PlayController {
public DeferredResult<WVPResult<StreamContent>> play(HttpServletRequest request, @PathVariable String deviceId, public DeferredResult<WVPResult<StreamContent>> play(HttpServletRequest request, @PathVariable String deviceId,
@PathVariable String channelId) { @PathVariable String channelId) {
logger.info("[开始点播] deviceId：{}, channelId：{}, ", deviceId, channelId);
// 获取可用的zlm // 获取可用的zlm
Device device = storager.queryVideoDevice(deviceId); Device device = storager.queryVideoDevice(deviceId);
MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
@ -109,13 +110,15 @@ public class PlayController {
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
result.onTimeout(()->{ result.onTimeout(()->{
logger.info("点播接口等待超时"); logger.info("[点播等待超时] deviceId：{}, channelId：{}, ", deviceId, channelId);
// 释放rtpserver // 释放rtpserver
WVPResult<StreamInfo> wvpResult = new WVPResult<>(); WVPResult<StreamInfo> wvpResult = new WVPResult<>();
wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setCode(ErrorCode.ERROR100.getCode());
wvpResult.setMsg("点播超时"); wvpResult.setMsg("点播超时");
requestMessage.setData(wvpResult); requestMessage.setData(wvpResult);
resultHolder.invokeResult(requestMessage); resultHolder.invokeAllResult(requestMessage);
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
storager.stopPlay(deviceId, channelId);
}); });
// 录像查询以channelId作为deviceId查询 // 录像查询以channelId作为deviceId查询
@ -168,7 +171,7 @@ public class PlayController {
} }
if (InviteSessionStatus.ok == inviteInfo.getStatus()) { if (InviteSessionStatus.ok == inviteInfo.getStatus()) {
try { try {
logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId); logger.info("[停止点播] {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null); cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage()); logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
@ -338,7 +341,7 @@ public class PlayController {
message.setKey(key); message.setKey(key);
message.setId(uuid); message.setId(uuid);
String fileName = deviceId + "_" + channelId + "_" + DateUtil.getNowForUrl() + "jpg"; String fileName = deviceId + "_" + channelId + "_" + DateUtil.getNowForUrl() + ".jpg";
playService.getSnap(deviceId, channelId, fileName, (code, msg, data) -> { playService.getSnap(deviceId, channelId, fileName, (code, msg, data) -> {
if (code == InviteErrorCode.SUCCESS.getCode()) { if (code == InviteErrorCode.SUCCESS.getCode()) {
message.setData(data); message.setData(data);

4
src/main/java/com/genersoft/iot/vmp/vmanager/log/LogController.java
View File

@ -42,7 +42,7 @@ public class LogController {
* @return * @return
*/ */
@GetMapping("/all") @GetMapping("/all")
@Operation(summary = "分页查询报警") @Operation(summary = "分页查询日志")
@Parameter(name = "query", description = "查询内容", required = true) @Parameter(name = "query", description = "查询内容", required = true)
@Parameter(name = "page", description = "当前页", required = true) @Parameter(name = "page", description = "当前页", required = true)
@Parameter(name = "count", description = "每页查询数量", required = true) @Parameter(name = "count", description = "每页查询数量", required = true)
@ -84,7 +84,7 @@ public class LogController {
* 清空日志 * 清空日志
* *
*/ */
@Operation(summary = "停止视频回放") @Operation(summary = "清空日志")
@DeleteMapping("/clear") @DeleteMapping("/clear")
public void clear() { public void clear() {
logService.clear(); logService.clear();

8
src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java
View File

@ -266,12 +266,4 @@ public class ServerController {
return result; return result;
} }
@PostMapping(value = "/test/getPort")
@ResponseBody
public int getPort() {
int result = sendRtpPortManager.getNextPort(mediaServerService.getDefaultMediaServer());
System.out.println(result);
return result;
}
} }

24
src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java
View File

@ -133,17 +133,29 @@ public class ApiStreamController {
result.put("ChannelName", deviceChannel.getName()); result.put("ChannelName", deviceChannel.getName());
result.put("ChannelCustomName", ""); result.put("ChannelCustomName", "");
result.put("FLV", inviteInfo.getStreamInfo().getFlv().getUrl()); result.put("FLV", inviteInfo.getStreamInfo().getFlv().getUrl());
result.put("HTTPS_FLV", inviteInfo.getStreamInfo().getHttps_flv().getUrl()); if(inviteInfo.getStreamInfo().getHttps_flv() != null) {
result.put("HTTPS_FLV", inviteInfo.getStreamInfo().getHttps_flv().getUrl());
}
result.put("WS_FLV", inviteInfo.getStreamInfo().getWs_flv().getUrl()); result.put("WS_FLV", inviteInfo.getStreamInfo().getWs_flv().getUrl());
result.put("WSS_FLV", inviteInfo.getStreamInfo().getWss_flv().getUrl()); if(inviteInfo.getStreamInfo().getWss_flv() != null) {
result.put("WSS_FLV", inviteInfo.getStreamInfo().getWss_flv().getUrl());
}
result.put("RTMP", inviteInfo.getStreamInfo().getRtmp().getUrl()); result.put("RTMP", inviteInfo.getStreamInfo().getRtmp().getUrl());
result.put("RTMPS", inviteInfo.getStreamInfo().getRtmps().getUrl()); if (inviteInfo.getStreamInfo().getRtmps() != null) {
result.put("RTMPS", inviteInfo.getStreamInfo().getRtmps().getUrl());
}
result.put("HLS", inviteInfo.getStreamInfo().getHls().getUrl()); result.put("HLS", inviteInfo.getStreamInfo().getHls().getUrl());
result.put("HTTPS_HLS", inviteInfo.getStreamInfo().getHttps_hls().getUrl()); if (inviteInfo.getStreamInfo().getHttps_hls() != null) {
result.put("HTTPS_HLS", inviteInfo.getStreamInfo().getHttps_hls().getUrl());
}
result.put("RTSP", inviteInfo.getStreamInfo().getRtsp().getUrl()); result.put("RTSP", inviteInfo.getStreamInfo().getRtsp().getUrl());
result.put("RTSPS", inviteInfo.getStreamInfo().getRtsps().getUrl()); if (inviteInfo.getStreamInfo().getRtsps() != null) {
result.put("RTSPS", inviteInfo.getStreamInfo().getRtsps().getUrl());
}
result.put("WEBRTC", inviteInfo.getStreamInfo().getRtc().getUrl()); result.put("WEBRTC", inviteInfo.getStreamInfo().getRtc().getUrl());
result.put("HTTPS_WEBRTC", inviteInfo.getStreamInfo().getRtcs().getUrl()); if (inviteInfo.getStreamInfo().getRtcs() != null) {
result.put("HTTPS_WEBRTC", inviteInfo.getStreamInfo().getRtcs().getUrl());
}
result.put("CDN", ""); result.put("CDN", "");
result.put("SnapURL", ""); result.put("SnapURL", "");
result.put("Transport", device.getTransport()); result.put("Transport", device.getTransport());

3
src/main/resources/all-application.yml
View File

@ -187,9 +187,6 @@ user-settings:
stream-on-demand: true stream-on-demand: true
# 推流鉴权， 默认开启 # 推流鉴权， 默认开启
push-authority: true push-authority: true
# 国标级联发流严格模式，严格模式会使用与sdp信息中一致的端口发流，端口共享media.rtp.port-range，这会损失一些性能，
# 非严格模式使用随机端口发流，性能更好， 默认关闭
gb-send-stream-strict: false
# 设备上线时是否自动同步通道 # 设备上线时是否自动同步通道
sync-channel-on-device-online: false sync-channel-on-device-online: false
# 国标级联语音喊话发流模式 * UDP:udp传输 TCP-ACTIVE：tcp主动模式 TCP-PASSIVE：tcp被动模式 # 国标级联语音喊话发流模式 * UDP:udp传输 TCP-ACTIVE：tcp主动模式 TCP-PASSIVE：tcp被动模式