diff --git a/doc/_content/introduction/config.md b/doc/_content/introduction/config.md index 9e8ac0f6..ad88140e 100644 --- a/doc/_content/introduction/config.md +++ b/doc/_content/introduction/config.md @@ -160,4 +160,4 @@ user-settings: 如果配置信息无误，你可以启动zlm，再启动wvp来测试了，启动成功的话，你可以在wvp的日志下看到zlm已连接的提示。 -接下来[部署到服务器](./_content/introduction/deployment.md)， 如何你只是本地运行直接再本地运行即可。 +接下来[部署到服务器](./_content/introduction/deployment.md)， 如果你只是本地运行直接在本地运行即可。 diff --git a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java index 041d7388..873feab5 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java @@ -1,5 +1,6 @@ package com.genersoft.iot.vmp.conf; +import org.apache.commons.lang3.ObjectUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Scheduled; @@ -45,6 +46,9 @@ public class DynamicTask { * @return */ public void startCron(String key, Runnable task, int cycleForCatalog) { + if(ObjectUtils.isEmpty(key)) { + return; + } ScheduledFuture future = futureMap.get(key); if (future != null) { if (future.isCancelled()) { @@ -73,6 +77,9 @@ public class DynamicTask { * @return */ public void startDelay(String key, Runnable task, int delay) { + if(ObjectUtils.isEmpty(key)) { + return; + } stop(key); // 获取执行的时刻 @@ -99,9 +106,12 @@ public class DynamicTask { } public boolean stop(String key) { + if(ObjectUtils.isEmpty(key)) { + return false; + } boolean result = false; - if (futureMap.get(key) != null && !futureMap.get(key).isCancelled() && !futureMap.get(key).isDone()) { - result = futureMap.get(key).cancel(false); + if (!ObjectUtils.isEmpty(futureMap.get(key)) && !futureMap.get(key).isCancelled() && !futureMap.get(key).isDone()) { + result = futureMap.get(key).cancel(true); futureMap.remove(key); runnableMap.remove(key); } @@ -109,6 +119,9 @@ public class DynamicTask { } public boolean contains(String key) { + if(ObjectUtils.isEmpty(key)) { + return false; + } return futureMap.get(key) != null; } @@ -117,6 +130,9 @@ public class DynamicTask { } public Runnable get(String key) { + if(ObjectUtils.isEmpty(key)) { + return null; + } return runnableMap.get(key); } diff --git a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java index 562c8640..0993d74a 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java @@ -43,8 +43,6 @@ public class UserSetting { private Boolean pushAuthority = Boolean.TRUE; - private Boolean gbSendStreamStrict = Boolean.FALSE; - private Boolean syncChannelOnDeviceOnline = Boolean.FALSE; private Boolean sipLog = Boolean.FALSE; @@ -208,14 +206,6 @@ public class UserSetting { this.pushAuthority = pushAuthority; } - public Boolean getGbSendStreamStrict() { - return gbSendStreamStrict; - } - - public void setGbSendStreamStrict(Boolean gbSendStreamStrict) { - this.gbSendStreamStrict = gbSendStreamStrict; - } - public Boolean getSyncChannelOnDeviceOnline() { return syncChannelOnDeviceOnline; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java index 541c111f..9bfab634 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java @@ -165,7 +165,7 @@ public class Device { * 是否开启ssrc校验，默认关闭，开启可以防止串流 */ @Schema(description = "是否开启ssrc校验，默认关闭，开启可以防止串流") - private boolean ssrcCheck = true; + private boolean ssrcCheck = false; /** * 地理坐标系， 目前支持 WGS84,GCJ02 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 d7dbe94f..80ea8909 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 @@ -333,7 +333,7 @@ public interface ISIPCommander { * @param endTime 报警发生终止时间（可选） * @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; /** * 订阅、取消订阅目录信息 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java index 3ea5378a..9d8ce364 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java @@ -167,6 +167,7 @@ public class SIPRequestHeaderPlarformProvider { public Request createMessageRequest(ParentPlatform parentPlatform, String content, SendRtpItem sendRtpItem) throws PeerUnavailableException, ParseException, InvalidArgumentException { CallIdHeader callIdHeader = SipFactory.getInstance().createHeaderFactory().createCallIdHeader(sendRtpItem.getCallId()); + callIdHeader.setCallId(sendRtpItem.getCallId()); return createMessageRequest(parentPlatform, content, sendRtpItem.getToTag(), SipUtils.getNewViaTag(), sendRtpItem.getFromTag(), callIdHeader); } 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 c1e72cd4..28d68013 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 @@ -1275,7 +1275,7 @@ public class SIPCommander implements ISIPCommander { * @return true = 命令发送成功 */ @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); String charset = device.getCharset(); @@ -1293,9 +1293,6 @@ public class SIPCommander implements ISIPCommander { if (!ObjectUtils.isEmpty(alarmMethod)) { cmdXml.append("" + alarmMethod + "\r\n"); } - if (!ObjectUtils.isEmpty(alarmType)) { - cmdXml.append("" + alarmType + "\r\n"); - } if (!ObjectUtils.isEmpty(startTime)) { cmdXml.append("" + startTime + "\r\n"); } 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 7dd23127..ab770c1f 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 @@ -98,15 +98,20 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In logger.warn("[收到ACK]：未找到来自{}，目标为({})的推流信息",fromUserId, toUserId); return; } - // tcp主动时，此时是级联下级平台，在回复200ok时，本地已经请求zlm开启监听，跳过下面步骤 - if (sendRtpItem.isTcpActive()) { - return; - } - logger.info("[收到ACK]：rtp/{}开始级推流, 目标={}:{}，SSRC={}, RTCP={}", sendRtpItem.getStream(), - sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp()); - // 取消设置的超时任务 - dynamicTask.stop(callIdHeader.getCallId()); + // tcp主动时，此时是级联下级平台，在回复200ok时，本地已经请求zlm开启监听，跳过下面步骤 + if (sendRtpItem.isTcpActive()) { + logger.info("收到ACK，rtp/{} TCP主动方式后续处理", sendRtpItem.getStreamId()); + return; + } + String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; 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); if (parentPlatform != null) { 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 7ae07a72..f999cf2c 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 @@ -148,7 +148,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId); } try { - logger.warn("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); + logger.info("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null); } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { 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 1f71ee0a..ff855de7 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 @@ -469,8 +469,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements sendRtpItem.setApp("rtp"); if ("Playback".equalsIgnoreCase(sessionName)) { sendRtpItem.setPlayType(InviteStreamType.PLAYBACK); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam()); - sendRtpItem.setStream(ssrcInfo.getStream()); + String startTimeStr = DateUtil.urlFormatter.format(start); + String endTimeStr = DateUtil.urlFormatter.format(end); + String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr; + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); // 写入redis， 超时时回复 redisCatchStorage.updateSendRTPSever(sendRtpItem); 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); - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - }else { - streamId = String.format("%08x", Integer.parseInt(ssrcInfo.getSsrc())).toUpperCase(); - } + String streamId = String.format("%s_%s", device.getDeviceId(), channelId); sendRtpItem.setStream(streamId); sendRtpItem.setSsrc(ssrcInfo.getSsrc()); redisCatchStorage.updateSendRTPSever(sendRtpItem); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java index 5c577ba6..7b9f69a3 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java @@ -33,7 +33,7 @@ import java.text.ParseException; 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"; @Autowired @@ -59,14 +59,19 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp // 未注册的设备不做处理 return; } - logger.info("[收到心跳]， device: {}", device.getDeviceId()); SIPRequest request = (SIPRequest) evt.getRequest(); + logger.info("[收到心跳]， device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId()); + // 回复200 OK try { responseAck(request, Response.OK); } catch (SipException | InvalidArgumentException | ParseException e) { 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()); if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) { @@ -80,7 +85,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp }else { long lastTime = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(device.getKeepaliveTime()); if (System.currentTimeMillis()/1000-lastTime > 10) { - device.setKeepaliveIntervalTime(new Long(System.currentTimeMillis()/1000-lastTime).intValue()); + device.setKeepaliveIntervalTime(Long.valueOf(System.currentTimeMillis()/1000-lastTime).intValue()); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java b/src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java index fbc1ee9e..c6460588 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java @@ -281,6 +281,6 @@ public class SipUtils { return null; } } - return localDateTime.format(DateUtil.formatterISO8601); + return localDateTime.format(DateUtil.formatter); } } \ No newline at end of file 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 c31c8b39..63a54dd4 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 @@ -200,7 +200,10 @@ public class ZLMHttpHookListener { String mediaServerId = json.getString("mediaServerId"); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); - + if (mediaInfo == null) { + return new HookResultForOnPublish(200, "success"); + } + // 推流鉴权的处理 if (!"rtp".equals(param.getApp())) { if (userSetting.getPushAuthority()) { // 推流鉴权 @@ -252,11 +255,21 @@ public class ZLMHttpHookListener { } }); + // 是否录像 if ("rtp".equals(param.getApp())) { result.setEnable_mp4(userSetting.getRecordSip()); } else { result.setEnable_mp4(userSetting.isRecordPushLive()); } + // 替换流地址 + if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) { + String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));; + InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc); + if (inviteInfo != null) { + result.setStream_replace(inviteInfo.getStream()); + logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream()); + } + } List ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream()); if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) { String deviceId = ssrcTransactionForAll.get(0).getDeviceId(); @@ -569,6 +582,7 @@ public class ZLMHttpHookListener { Device device = deviceService.getDevice(inviteInfo.getDeviceId()); if (device != null) { try { + // 多查询一次防止已经被处理了 InviteInfo info = inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); if (info != null) { @@ -643,7 +657,7 @@ public class ZLMHttpHookListener { if ("rtp".equals(param.getApp())) { String[] s = param.getStream().split("_"); - if (!mediaInfo.isRtpEnable() || (s.length != 2 && s.length != 4)) { + if ((s.length != 2 && s.length != 4)) { defaultResult.setResult(HookResult.SUCCESS()); return defaultResult; } @@ -672,7 +686,6 @@ public class ZLMHttpHookListener { result.onTimeout(() -> { logger.info("[ZLM HOOK] 预览流自动点播, 等待超时"); - // 释放rtpserver msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时")); resultHolder.invokeResult(msg); }); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java index 81cae37c..72fed079 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java @@ -168,13 +168,9 @@ public class ZLMServerFactory { public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String deviceId, String channelId, boolean tcp, boolean rtcp){ - // 默认为随机端口 - int localPort = 0; - if (userSetting.getGbSendStreamStrict()) { - localPort = sendRtpPortManager.getNextPort(serverItem); - if (localPort == 0) { - return null; - } + int localPort = sendRtpPortManager.getNextPort(serverItem); + if (localPort == 0) { + return null; } SendRtpItem sendRtpItem = new SendRtpItem(); sendRtpItem.setIp(ip); @@ -204,13 +200,10 @@ public class ZLMServerFactory { */ public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String app, String stream, String channelId, boolean tcp, boolean rtcp){ - // 默认为随机端口 - int localPort = 0; - if (userSetting.getGbSendStreamStrict()) { - localPort = sendRtpPortManager.getNextPort(serverItem); - if (localPort == 0) { - return null; - } + + int localPort = sendRtpPortManager.getNextPort(serverItem); + if (localPort == 0) { + return null; } SendRtpItem sendRtpItem = new SendRtpItem(); sendRtpItem.setIp(ip); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java index 68d969f4..12d83627 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java @@ -6,6 +6,7 @@ public class HookResultForOnPublish extends HookResult{ private boolean enable_mp4; private int mp4_max_second; private String mp4_save_path; + private String stream_replace; public HookResultForOnPublish() { } @@ -51,12 +52,21 @@ public class HookResultForOnPublish extends HookResult{ this.mp4_save_path = mp4_save_path; } + public String getStream_replace() { + return stream_replace; + } + + public void setStream_replace(String stream_replace) { + this.stream_replace = stream_replace; + } + @Override public String toString() { return "HookResultForOnPublish{" + "enable_audio=" + enable_audio + ", enable_mp4=" + enable_mp4 + ", mp4_max_second=" + mp4_max_second + + ", stream_replace=" + stream_replace + ", mp4_save_path='" + mp4_save_path + '\'' + '}'; } diff --git a/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java b/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java index ae30f26e..c06400dc 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java @@ -74,5 +74,13 @@ public interface IInviteStreamService { int getStreamInfoCount(String mediaServerId); + /** + * 获取MediaServer下的流信息 + */ + InviteInfo getInviteInfoBySSRC(String ssrc); + /** + * 更新ssrc + */ + InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrcInResponse); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index 0fa3aa61..5192d1d0 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -190,6 +190,17 @@ public class DeviceServiceImpl implements IDeviceService { 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 diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java index f4128163..d630a2c0 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java @@ -77,10 +77,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService { } String key = VideoManagerConstants.INVITE_PREFIX + - "_" + inviteInfoForUpdate.getType() + - "_" + inviteInfoForUpdate.getDeviceId() + - "_" + inviteInfoForUpdate.getChannelId() + - "_" + inviteInfoForUpdate.getStream(); + ":" + inviteInfoForUpdate.getType() + + ":" + inviteInfoForUpdate.getDeviceId() + + ":" + inviteInfoForUpdate.getChannelId() + + ":" + inviteInfoForUpdate.getStream()+ + ":" + inviteInfoForUpdate.getSsrcInfo().getSsrc(); redisTemplate.opsForValue().set(key, inviteInfoForUpdate); } @@ -93,11 +94,15 @@ public class InviteStreamServiceImpl implements IInviteStreamService { } removeInviteInfo(inviteInfoInDb); String key = VideoManagerConstants.INVITE_PREFIX + - "_" + inviteInfo.getType() + - "_" + inviteInfo.getDeviceId() + - "_" + inviteInfo.getChannelId() + - "_" + stream; + ":" + inviteInfo.getType() + + ":" + inviteInfo.getDeviceId() + + ":" + inviteInfo.getChannelId() + + ":" + stream + + ":" + inviteInfo.getSsrcInfo().getSsrc(); inviteInfoInDb.setStream(stream); + if (inviteInfoInDb.getSsrcInfo() != null) { + inviteInfoInDb.getSsrcInfo().setStream(stream); + } redisTemplate.opsForValue().set(key, inviteInfoInDb); return inviteInfoInDb; } @@ -105,10 +110,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService { @Override public InviteInfo getInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) { String key = VideoManagerConstants.INVITE_PREFIX + - "_" + (type != null ? type : "*") + - "_" + (deviceId != null ? deviceId : "*") + - "_" + (channelId != null ? channelId : "*") + - "_" + (stream != null ? stream : "*"); + ":" + (type != null ? type : "*") + + ":" + (deviceId != null ? deviceId : "*") + + ":" + (channelId != null ? channelId : "*") + + ":" + (stream != null ? stream : "*") + + ":*"; List scanResult = RedisUtil.scan(redisTemplate, key); if (scanResult.size() != 1) { return null; @@ -130,10 +136,11 @@ public class InviteStreamServiceImpl implements IInviteStreamService { @Override public void removeInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) { String scanKey = VideoManagerConstants.INVITE_PREFIX + - "_" + (type != null ? type : "*") + - "_" + (deviceId != null ? deviceId : "*") + - "_" + (channelId != null ? channelId : "*") + - "_" + (stream != null ? stream : "*"); + ":" + (type != null ? type : "*") + + ":" + (deviceId != null ? deviceId : "*") + + ":" + (channelId != null ? channelId : "*") + + ":" + (stream != null ? stream : "*") + + ":*"; List scanResult = RedisUtil.scan(redisTemplate, scanKey); if (scanResult.size() > 0) { for (Object keyObj : scanResult) { @@ -171,10 +178,10 @@ public class InviteStreamServiceImpl implements IInviteStreamService { } 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 if (stream != null) { - key += ("_" + stream); + key += (":" + stream); } return key; } @@ -188,7 +195,7 @@ public class InviteStreamServiceImpl implements IInviteStreamService { @Override public int getStreamInfoCount(String mediaServerId) { int count = 0; - String key = VideoManagerConstants.INVITE_PREFIX + "_*_*_*_*"; + String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:*"; List scanResult = RedisUtil.scan(redisTemplate, key); if (scanResult.size() == 0) { return 0; @@ -219,11 +226,42 @@ public class InviteStreamServiceImpl implements IInviteStreamService { 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 if (stream != null) { - key += ("_" + stream); + key += (":" + stream); } return key; } + + @Override + public InviteInfo getInviteInfoBySSRC(String ssrc) { + String key = VideoManagerConstants.INVITE_PREFIX + ":*:*:*:*:" + ssrc; + List scanResult = RedisUtil.scan(redisTemplate, key); + if (scanResult.size() != 1) { + return null; + } + + return (InviteInfo) redisTemplate.opsForValue().get(scanResult.get(0)); + } + + @Override + public InviteInfo updateInviteInfoForSSRC(InviteInfo inviteInfo, String ssrc) { + InviteInfo inviteInfoInDb = getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); + if (inviteInfoInDb == null) { + return null; + } + removeInviteInfo(inviteInfoInDb); + String key = VideoManagerConstants.INVITE_PREFIX + + ":" + inviteInfo.getType() + + ":" + inviteInfo.getDeviceId() + + ":" + inviteInfo.getChannelId() + + ":" + inviteInfo.getStream() + + ":" + inviteInfo.getSsrcInfo().getSsrc(); + if (inviteInfoInDb.getSsrcInfo() != null) { + inviteInfoInDb.getSsrcInfo().setSsrc(ssrc); + } + redisTemplate.opsForValue().set(key, inviteInfoInDb); + return inviteInfoInDb; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index 7a523632..b5faf1bf 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 @@ -153,9 +153,14 @@ public class MediaServerServiceImpl implements IMediaServerService { if (streamId == null) { 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; 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 { rtpServerPort = mediaServerItem.getRtpProxyPort(); } 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 de32e5ea..cb846e6f 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 @@ -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.transmit.cmd.ISIPCommanderForPlatform; 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.media.zlm.*; 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.gb28181.play.bean.AudioBroadcastEvent; import gov.nist.javax.sip.message.SIPResponse; +import gov.nist.javax.sip.message.SIPResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -147,15 +150,21 @@ public class PlayServiceImpl implements IPlayService { @Override public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, String ssrc, ErrorCallback callback) { if (mediaServerItem == null) { + logger.warn("[点播] 未找到可用的zlm deviceId: {},channelId:{}", deviceId, channelId); throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm"); } 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); if (inviteInfo != null ) { if (inviteInfo.getStreamInfo() == null) { // 点播发起了但是尚未成功, 仅注册回调等待结果即可 inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); + logger.info("[点播开始] 已经请求中，等待结果， deviceId: {}, channelId: {}", device.getDeviceId(), channelId); return inviteInfo.getSsrcInfo(); }else { StreamInfo streamInfo = inviteInfo.getStreamInfo(); @@ -178,6 +187,7 @@ public class PlayServiceImpl implements IPlayService { InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); + logger.info("[点播已存在] 直接返回， deviceId: {}, channelId: {}", device.getDeviceId(), channelId); return inviteInfo.getSsrcInfo(); }else { // 点播发起了但是尚未成功, 仅注册回调等待结果即可 @@ -187,11 +197,8 @@ public class PlayServiceImpl implements IPlayService { } } } - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, false,device.getStreamModeForParam()); + String streamId = String.format("%s_%s", device.getDeviceId(), channelId);; + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); if (ssrcInfo == null) { callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null); inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, @@ -200,7 +207,6 @@ public class PlayServiceImpl implements IPlayService { null); return null; } - // TODO 记录点播的状态 play(mediaServerItem, ssrcInfo, device, channelId, callback); return ssrcInfo; } @@ -357,8 +363,8 @@ public class PlayServiceImpl implements IPlayService { null); return; } - logger.info("[点播开始] deviceId: {}, channelId: {},码流类型：{}, 收流端口： {}, 收流模式：{}, SSRC: {}, SSRC校验：{}", - device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流", ssrcInfo.getPort(), + logger.info("[点播开始] deviceId: {}, channelId: {},码流类型：{}, 收流端口： {}, STREAM：{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", + device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流", ssrcInfo.getPort(), ssrcInfo.getStream(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); //端口获取失败的ssrcInfo 没有必要发送点播指令 if (ssrcInfo.getPort() <= 0) { @@ -389,16 +395,6 @@ public class PlayServiceImpl implements IPlayService { device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流", 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); inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, 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, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流"); snapOnPlay(mediaServerItemInuse, device.getDeviceId(), channelId, ssrcInfo.getStream()); - }, (event) -> { - inviteInfo.setStatus(InviteSessionStatus.ok); - - ResponseEvent responseEvent = (ResponseEvent) event.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).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); + }, (eventResult) -> { + // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题 + InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId, + timeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAY); }, (event) -> { dynamicTask.stop(timeOutTaskKey); 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 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 callback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { - return; + logger.warn("[录像回放] 未找到设备 deviceId: {},channelId:{}", deviceId, channelId); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到设备：" + deviceId); } + MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - String stream = null; - if (newMediaServerItem.isRtpEnable()) { - String startTimeStr = startTime.replace("-", "") - .replace(":", "") - .replace(" ", ""); - System.out.println(startTimeStr); - String endTimeTimeStr = endTime.replace("-", "") - .replace(":", "") - .replace(" ", ""); - System.out.println(endTimeTimeStr); - stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr; + if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE") && ! newMediaServerItem.isRtpEnable()) { + logger.warn("[录像回放] 单端口收流时不支持TCP主动方式收流 deviceId: {},channelId:{}", deviceId, channelId); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "单端口收流时不支持TCP主动方式收流"); } - 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); } @@ -795,113 +715,13 @@ public class PlayServiceImpl implements IPlayService { try { cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, hookEvent, eventResult -> { - inviteInfo.setStatus(InviteSessionStatus.ok); - 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)) { - 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(); + // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题 + InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId, + playBackTimeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAYBACK); - 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); } catch (InvalidArgumentException | SipException | ParseException e) { - logger.error("[命令发送失败] 回放: {}", e.getMessage()); + logger.error("[命令发送失败] 录像回放: {}", e.getMessage()); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); 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 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 public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, ErrorCallback callback) { Device device = storager.queryVideoDevice(deviceId); @@ -925,6 +829,7 @@ public class PlayServiceImpl implements IPlayService { null); return; } + // 录像下载不使用固定流地址，固定流地址会导致如果开始时间与结束时间一致时文件错误的叠加在一起 SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,false, device.getStreamModeForParam()); download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, callback); } @@ -993,108 +898,9 @@ public class PlayServiceImpl implements IPlayService { try { cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, hookEvent, errorEvent, eventResult ->{ - inviteInfo.setStatus(InviteSessionStatus.ok); - 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)) { - 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); + // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题 + InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId, + downLoadTimeOutTaskKey, callback, inviteInfo, InviteSessionType.DOWNLOAD); }); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 录像下载: {}", e.getMessage()); diff --git a/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java b/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java index 923f8346..23cb9dac 100644 --- a/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java +++ b/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java @@ -7,6 +7,7 @@ import java.time.LocalDateTime; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; +import java.time.temporal.ChronoUnit; import java.time.temporal.TemporalAccessor; import java.util.Locale; @@ -106,4 +107,9 @@ public class DateUtil { LocalDateTime nowDateTime = LocalDateTime.now(); return formatterISO8601.format(nowDateTime); } + + public static long getDifferenceForNow(String keepaliveTime) { + Instant beforeInstant = Instant.from(formatter.parse(keepaliveTime)); + return ChronoUnit.MILLIS.between(beforeInstant, Instant.now()); + } } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java index 8e0e2012..0b23a9c0 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java @@ -67,7 +67,7 @@ public class MediaController { && streamAuthorityInfo.getCallId().equals(callId)) { authority = true; }else { - throw new ControllerException(ErrorCode.ERROR400); + throw new ControllerException(ErrorCode.ERROR400.getCode(), "获取播放地址鉴权失败"); } }else { // 是否登陆用户, 登陆用户返回完整信息 diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java index e67e373d..97017697 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java @@ -97,6 +97,7 @@ public class PlayController { public DeferredResult> play(HttpServletRequest request, @PathVariable String deviceId, @PathVariable String channelId) { + logger.info("[开始点播] deviceId：{}, channelId：{}, ", deviceId, channelId); // 获取可用的zlm Device device = storager.queryVideoDevice(deviceId); MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); @@ -109,13 +110,15 @@ public class PlayController { DeferredResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); result.onTimeout(()->{ - logger.info("点播接口等待超时"); + logger.info("[点播等待超时] deviceId：{}, channelId：{}, ", deviceId, channelId); // 释放rtpserver WVPResult wvpResult = new WVPResult<>(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播超时"); requestMessage.setData(wvpResult); - resultHolder.invokeResult(requestMessage); + resultHolder.invokeAllResult(requestMessage); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + storager.stopPlay(deviceId, channelId); }); // 录像查询以channelId作为deviceId查询 @@ -168,7 +171,7 @@ public class PlayController { } if (InviteSessionStatus.ok == inviteInfo.getStatus()) { try { - logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId); + logger.info("[停止点播] {}/{}", device.getDeviceId(), channelId); cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null); } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage()); @@ -338,7 +341,7 @@ public class PlayController { message.setKey(key); 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) -> { if (code == InviteErrorCode.SUCCESS.getCode()) { message.setData(data); diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/log/LogController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/log/LogController.java index a412b104..3aabd924 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/log/LogController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/log/LogController.java @@ -42,7 +42,7 @@ public class LogController { * @return */ @GetMapping("/all") - @Operation(summary = "分页查询报警") + @Operation(summary = "分页查询日志") @Parameter(name = "query", description = "查询内容", required = true) @Parameter(name = "page", description = "当前页", required = true) @Parameter(name = "count", description = "每页查询数量", required = true) @@ -84,7 +84,7 @@ public class LogController { * 清空日志 * */ - @Operation(summary = "停止视频回放") + @Operation(summary = "清空日志") @DeleteMapping("/clear") public void clear() { logService.clear(); diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java index f4bcb1a9..72641be4 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java @@ -266,12 +266,4 @@ public class ServerController { return result; } - - @PostMapping(value = "/test/getPort") - @ResponseBody - public int getPort() { - int result = sendRtpPortManager.getNextPort(mediaServerService.getDefaultMediaServer()); - System.out.println(result); - return result; - } } diff --git a/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java b/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java index 39f6388f..ddf677ce 100644 --- a/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java +++ b/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java @@ -133,17 +133,29 @@ public class ApiStreamController { result.put("ChannelName", deviceChannel.getName()); result.put("ChannelCustomName", ""); 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("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("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("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("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("HTTPS_WEBRTC", inviteInfo.getStreamInfo().getRtcs().getUrl()); + if (inviteInfo.getStreamInfo().getRtcs() != null) { + result.put("HTTPS_WEBRTC", inviteInfo.getStreamInfo().getRtcs().getUrl()); + } result.put("CDN", ""); result.put("SnapURL", ""); result.put("Transport", device.getTransport()); diff --git a/src/main/resources/all-application.yml b/src/main/resources/all-application.yml index 9b94f6f6..13689616 100644 --- a/src/main/resources/all-application.yml +++ b/src/main/resources/all-application.yml @@ -187,9 +187,6 @@ user-settings: stream-on-demand: true # 推流鉴权， 默认开启 push-authority: true - # 国标级联发流严格模式，严格模式会使用与sdp信息中一致的端口发流，端口共享media.rtp.port-range，这会损失一些性能， - # 非严格模式使用随机端口发流，性能更好， 默认关闭 - gb-send-stream-strict: false # 设备上线时是否自动同步通道 sync-channel-on-device-online: false # 国标级联语音喊话发流模式 * UDP:udp传输 TCP-ACTIVE：tcp主动模式 TCP-PASSIVE：tcp被动模式