diff --git a/doc/_content/introduction/config.md b/doc/_content/introduction/config.md index 9e8ac0f63..ad88140e9 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/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index a90670d87..130f49de4 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -158,7 +158,9 @@ public class VideoManagerConstants { public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_"; public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_"; public static final String WVP_OTHER_SEND_RTP_INFO = "VMP_OTHER_SEND_RTP_INFO_"; + public static final String WVP_OTHER_SEND_PS_INFO = "VMP_OTHER_SEND_PS_INFO_"; public static final String WVP_OTHER_RECEIVE_RTP_INFO = "VMP_OTHER_RECEIVE_RTP_INFO_"; + public static final String WVP_OTHER_RECEIVE_PS_INFO = "VMP_OTHER_RECEIVE_PS_INFO_"; /** * Redis Const 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 041d73884..873feab58 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/gb28181/bean/Device.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java index 1c7baac9c..34d624f4b 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/SIPRequestHeaderPlarformProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java index fe130a549..9a00c1799 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 @@ -166,6 +166,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/event/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java index 61a600410..d189048b7 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 @@ -111,9 +111,20 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In logger.warn("[收到ACK]：未找到通道({})的推流信息", channelId); return; } + // 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.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc()); + logger.info("收到ACK，rtp/{}开始向上级推流, 目标={}:{}，SSRC={}, 协议:{}", + sendRtpItem.getStreamId(), + sendRtpItem.getIp(), + sendRtpItem.getPort(), + sendRtpItem.getSsrc(), + sendRtpItem.isTcp()?(sendRtpItem.isTcpActive()?"TCP主动":"TCP被动"):"UDP" + ); Map param = new HashMap<>(12); param.put("vhost","__defaultVhost__"); param.put("app",sendRtpItem.getApp()); @@ -130,10 +141,6 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In // 开启rtcp保活 param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0"); } - // tcp主动时，此时是级联下级平台，在回复200ok时，本地已经请求zlm开启监听，跳过下面步骤 - if (sendRtpItem.isTcpActive()) { - return; - } if (mediaInfo == null) { RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java index 374e5dcc0..b6aac9c7e 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 @@ -138,7 +138,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 397a4a0f7..d2d9657b0 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 @@ -459,7 +459,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, device.getStreamModeForParam()); + 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()); sendRtpItem.setStreamId(ssrcInfo.getStream()); // 写入redis， 超时时回复 redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -520,12 +523,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.setStreamId(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 5c577ba68..7b9f69a3f 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/media/zlm/SendRtpPortManager.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java index f960c7dc8..3f28d02ad 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java @@ -30,7 +30,7 @@ public class SendRtpPortManager { private final String KEY = "VM_MEDIA_SEND_RTP_PORT_"; - public int getNextPort(MediaServerItem mediaServer) { + public synchronized int getNextPort(MediaServerItem mediaServer) { if (mediaServer == null) { logger.warn("[发送端口管理] 参数错误，mediaServer为NULL"); return -1; @@ -50,17 +50,15 @@ public class SendRtpPortManager { String sendRtpPortRange = mediaServer.getSendRtpPortRange(); int startPort; int endPort; - if (sendRtpPortRange == null) { - logger.warn("{}未设置发送端口默认值，自动使用40000-50000作为端口范围", mediaServer.getId()); + if (sendRtpPortRange != null) { String[] portArray = sendRtpPortRange.split(","); if (portArray.length != 2 || !NumberUtils.isParsable(portArray[0]) || !NumberUtils.isParsable(portArray[1])) { - logger.warn("{}发送端口配置格式错误，自动使用40000-50000作为端口范围", mediaServer.getId()); + logger.warn("{}发送端口配置格式错误，自动使用50000-60000作为端口范围", mediaServer.getId()); startPort = 50000; endPort = 60000; }else { - if ( Integer.parseInt(portArray[1]) - Integer.parseInt(portArray[0]) < 1) { - logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用40000-50000作为端口范围", mediaServer.getId()); + logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用50000-60000作为端口范围", mediaServer.getId()); startPort = 50000; endPort = 60000; }else { @@ -69,6 +67,7 @@ public class SendRtpPortManager { } } }else { + logger.warn("{}未设置发送端口默认值，自动使用50000-60000作为端口范围", mediaServer.getId()); startPort = 50000; endPort = 60000; } @@ -76,10 +75,35 @@ public class SendRtpPortManager { logger.warn("{}获取redis连接信息失败", mediaServer.getId()); return -1; } +// RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory()); +// return redisAtomicInteger.getAndUpdate((current)->{ +// return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort)); +// }); + return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap); + } + + private synchronized int getSendPort(int startPort, int endPort, String sendIndexKey, Map sendRtpItemMap){ RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory()); - return redisAtomicInteger.getAndUpdate((current)->{ - return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort)); - }); + if (redisAtomicInteger.get() < startPort) { + redisAtomicInteger.set(startPort); + return startPort; + }else { + int port = redisAtomicInteger.getAndIncrement(); + if (port > endPort) { + redisAtomicInteger.set(startPort); + if (sendRtpItemMap.containsKey(startPort)) { + return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap); + }else { + return startPort; + } + } + if (sendRtpItemMap.containsKey(port)) { + return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap); + }else { + return port; + } + } + } interface CheckPortCallback{ 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 9b2864fd4..3c8f2de21 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 @@ -29,6 +29,7 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo; import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo; import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import org.slf4j.Logger; @@ -193,7 +194,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()) { // 推流鉴权 @@ -245,11 +249,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(); @@ -285,7 +299,10 @@ public class ZLMHttpHookListener { if (param.getApp().equalsIgnoreCase("rtp")) { String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream(); OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(receiveKey); - if (otherRtpSendInfo != null) { + + String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + param.getStream(); + OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(receiveKeyForPS); + if (otherRtpSendInfo != null || otherPsSendInfo != null) { result.setEnable_mp4(true); } } @@ -494,6 +511,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) { @@ -561,7 +579,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; } @@ -590,7 +608,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/dto/hook/HookResultForOnPublish.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java index 68d969f41..12d83627c 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 ae30f26e7..c06400dce 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/InviteStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java index f41281633..d630a2c02 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 142b81006..580561b32 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 @@ -151,9 +151,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, reUsePort, tcpMode); + rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0)?Integer.parseInt(ssrc):0, port, 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 0c287a9af..f2653f70c 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 @@ -12,11 +12,11 @@ import com.genersoft.iot.vmp.conf.exception.ServiceException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; -import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; 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.media.zlm.AssistRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; @@ -34,10 +34,10 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import gov.nist.javax.sip.message.SIPResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import org.springframework.util.ObjectUtils; @@ -95,7 +95,6 @@ public class PlayServiceImpl implements IPlayService { @Autowired private VideoStreamSessionManager streamSession; - @Autowired private IDeviceService deviceService; @@ -108,25 +107,25 @@ public class PlayServiceImpl implements IPlayService { @Autowired private ZlmHttpHookSubscribe subscribe; - @Autowired - private SSRCFactory ssrcFactory; - - @Autowired - private RedisTemplate redisTemplate; - @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(); @@ -149,6 +148,7 @@ public class PlayServiceImpl implements IPlayService { InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); + logger.info("[点播已存在] 直接返回， deviceId: {}, channelId: {}", device.getDeviceId(), channelId); return inviteInfo.getSsrcInfo(); }else { // 点播发起了但是尚未成功, 仅注册回调等待结果即可 @@ -158,10 +158,7 @@ public class PlayServiceImpl implements IPlayService { } } } - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - } + 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); @@ -171,7 +168,6 @@ public class PlayServiceImpl implements IPlayService { null); return null; } - // TODO 记录点播的状态 play(mediaServerItem, ssrcInfo, device, channelId, callback); return ssrcInfo; } @@ -187,8 +183,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) { @@ -219,16 +215,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); @@ -272,128 +258,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()); @@ -430,6 +298,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); + } + } + /** * 点播成功时调用截图. * @@ -536,21 +445,22 @@ 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主动方式收流"); } + 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); } @@ -624,113 +534,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; @@ -741,6 +551,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) { @@ -755,6 +649,7 @@ public class PlayServiceImpl implements IPlayService { null); return; } + // 录像下载不使用固定流地址，固定流地址会导致如果开始时间与结束时间一致时文件错误的叠加在一起 SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, callback); } @@ -823,108 +718,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 923f8346f..23cb9dac9 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/bean/OtherPsSendInfo.java b/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java new file mode 100644 index 000000000..ac9840921 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java @@ -0,0 +1,137 @@ +package com.genersoft.iot.vmp.vmanager.bean; + +public class OtherPsSendInfo { + + /** + * 发流IP + */ + private String sendLocalIp; + + /** + * 发流端口 + */ + private int sendLocalPort; + + /** + * 收流IP + */ + private String receiveIp; + + /** + * 收流端口 + */ + private int receivePort; + + + /** + * 会话ID + */ + private String callId; + + /** + * 流ID + */ + private String stream; + + /** + * 推流应用名 + */ + private String pushApp; + + /** + * 推流流ID + */ + private String pushStream; + + /** + * 推流SSRC + */ + private String pushSSRC; + + public String getSendLocalIp() { + return sendLocalIp; + } + + public void setSendLocalIp(String sendLocalIp) { + this.sendLocalIp = sendLocalIp; + } + + public int getSendLocalPort() { + return sendLocalPort; + } + + public void setSendLocalPort(int sendLocalPort) { + this.sendLocalPort = sendLocalPort; + } + + public String getReceiveIp() { + return receiveIp; + } + + public void setReceiveIp(String receiveIp) { + this.receiveIp = receiveIp; + } + + public int getReceivePort() { + return receivePort; + } + + public void setReceivePort(int receivePort) { + this.receivePort = receivePort; + } + + public String getCallId() { + return callId; + } + + public void setCallId(String callId) { + this.callId = callId; + } + + public String getStream() { + return stream; + } + + public void setStream(String stream) { + this.stream = stream; + } + + public String getPushApp() { + return pushApp; + } + + public void setPushApp(String pushApp) { + this.pushApp = pushApp; + } + + public String getPushStream() { + return pushStream; + } + + public void setPushStream(String pushStream) { + this.pushStream = pushStream; + } + + public String getPushSSRC() { + return pushSSRC; + } + + public void setPushSSRC(String pushSSRC) { + this.pushSSRC = pushSSRC; + } + + @Override + public String toString() { + return "OtherPsSendInfo{" + + "sendLocalIp='" + sendLocalIp + '\'' + + ", sendLocalPort=" + sendLocalPort + + ", receiveIp='" + receiveIp + '\'' + + ", receivePort=" + receivePort + + ", callId='" + callId + '\'' + + ", stream='" + stream + '\'' + + ", pushApp='" + pushApp + '\'' + + ", pushStream='" + pushStream + '\'' + + ", pushSSRC='" + pushSSRC + '\'' + + '}'; + } +} 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 8e0e2012f..0b23a9c0d 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 8ae875246..3d233dcf8 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 @@ -92,6 +92,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); @@ -104,13 +105,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查询 @@ -163,7 +166,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()); diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java new file mode 100644 index 000000000..81ccb5734 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java @@ -0,0 +1,324 @@ +package com.genersoft.iot.vmp.vmanager.ps; + +import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.VideoManagerConstants; +import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.conf.UserSetting; +import com.genersoft.iot.vmp.conf.exception.ControllerException; +import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; +import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; +import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; +import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam; +import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.utils.redis.RedisUtil; +import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.web.bind.annotation.*; + +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +@SuppressWarnings("rawtypes") +@Tag(name = "第三方PS服务对接") + +@RestController +@RequestMapping("/api/ps") +public class PsController { + + private final static Logger logger = LoggerFactory.getLogger(PsController.class); + + @Autowired + private ZLMServerFactory zlmServerFactory; + + @Autowired + private ZlmHttpHookSubscribe hookSubscribe; + + @Autowired + private IMediaServerService mediaServerService; + + @Autowired + private SendRtpPortManager sendRtpPortManager; + + @Autowired + private UserSetting userSetting; + + @Autowired + private DynamicTask dynamicTask; + + + @Autowired + private RedisTemplate redisTemplate; + + + @GetMapping(value = "/receive/open") + @ResponseBody + @Operation(summary = "开启收流和获取发流信息") + @Parameter(name = "isSend", description = "是否发送，false时只开启收流， true同时返回推流信息", required = true) + @Parameter(name = "callId", description = "整个过程的唯一标识，为了与后续接口关联", required = true) + @Parameter(name = "ssrc", description = "来源流的SSRC，不传则不校验来源ssrc", required = false) + @Parameter(name = "stream", description = "形成的流的ID", required = true) + @Parameter(name = "tcpMode", description = "收流模式， 0为UDP， 1为TCP被动", required = true) + @Parameter(name = "callBack", description = "回调地址，如果收流超时会通道回调通知，回调为get请求，参数为callId", required = true) + public OtherPsSendInfo openRtpServer(Boolean isSend, @RequestParam(required = false)String ssrc, String callId, String stream, Integer tcpMode, String callBack) { + + logger.info("[第三方PS服务对接->开启收流和获取发流信息] isSend->{}, ssrc->{}, callId->{}, stream->{}, tcpMode->{}, callBack->{}", + isSend, ssrc, callId, stream, tcpMode==0?"UDP":"TCP被动", callBack); + + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + if (mediaServerItem == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"没有可用的MediaServer"); + } + if (stream == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"stream参数不可为空"); + } + if (isSend != null && isSend && callId == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"isSend为true时，CallID不能为空"); + } + int ssrcInt = 0; + if (ssrc != null) { + try { + ssrcInt = Integer.parseInt(ssrc); + }catch (NumberFormatException e) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"ssrc格式错误"); + } + } + String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + callId + "_" + stream; + int localPort = zlmServerFactory.createRTPServer(mediaServerItem, stream, ssrcInt, null, false, tcpMode); + if (localPort == 0) { + throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败"); + } + // 注册回调如果rtp收流超时则通过回调发送通知 + if (callBack != null) { + HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId()); + // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 + hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, + (mediaServerItemInUse, hookParam)->{ + OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam; + if (stream.equals(serverTimeoutHookParam.getStream_id())) { + logger.info("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId); + // 将信息写入redis中，以备后用 + redisTemplate.delete(receiveKey); + OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder(); + OkHttpClient client = httpClientBuilder.build(); + String url = callBack + "?callId=" + callId; + Request request = new Request.Builder().get().url(url).build(); + try { + client.newCall(request).execute(); + } catch (IOException e) { + logger.error("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e); + } + hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout); + } + }); + } + OtherPsSendInfo otherPsSendInfo = new OtherPsSendInfo(); + otherPsSendInfo.setReceiveIp(mediaServerItem.getSdpIp()); + otherPsSendInfo.setReceivePort(localPort); + otherPsSendInfo.setCallId(callId); + otherPsSendInfo.setStream(stream); + + // 将信息写入redis中，以备后用 + redisTemplate.opsForValue().set(receiveKey, otherPsSendInfo); + if (isSend != null && isSend) { + String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; + // 预创建发流信息 + int port = sendRtpPortManager.getNextPort(mediaServerItem); + + otherPsSendInfo.setSendLocalIp(mediaServerItem.getSdpIp()); + otherPsSendInfo.setSendLocalPort(port); + // 将信息写入redis中，以备后用 + redisTemplate.opsForValue().set(key, otherPsSendInfo, 300, TimeUnit.SECONDS); + logger.info("[第三方PS服务对接->开启收流和获取发流信息] 结果，callId->{}， {}", callId, otherPsSendInfo); + } + return otherPsSendInfo; + } + + @GetMapping(value = "/receive/close") + @ResponseBody + @Operation(summary = "关闭收流") + @Parameter(name = "stream", description = "流的ID", required = true) + public void closeRtpServer(String stream) { + logger.info("[第三方PS服务对接->关闭收流] stream->{}", stream); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + zlmServerFactory.closeRtpServer(mediaServerItem,stream); + String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream; + List scan = RedisUtil.scan(redisTemplate, receiveKey); + if (!scan.isEmpty()) { + for (Object key : scan) { + // 将信息写入redis中，以备后用 + redisTemplate.delete(key); + } + } + } + + @GetMapping(value = "/send/start") + @ResponseBody + @Operation(summary = "发送流") + @Parameter(name = "ssrc", description = "发送流的SSRC", required = true) + @Parameter(name = "dstIp", description = "目标收流IP", required = true) + @Parameter(name = "dstPort", description = "目标收流端口", required = true) + @Parameter(name = "app", description = "待发送应用名", required = true) + @Parameter(name = "stream", description = "待发送流Id", required = true) + @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true) + @Parameter(name = "isUdp", description = "是否为UDP", required = true) + public void sendRTP(String ssrc, + String dstIp, + Integer dstPort, + String app, + String stream, + String callId, + Boolean isUdp + ) { + logger.info("[第三方PS服务对接->发送流] " + + "ssrc->{}, \r\n" + + "dstIp->{}, \n" + + "dstPort->{}, \n" + + "app->{}, \n" + + "stream->{}, \n" + + "callId->{} \n", + ssrc, + dstIp, + dstPort, + app, + stream, + callId); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; + OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key); + if (sendInfo == null) { + sendInfo = new OtherPsSendInfo(); + } + sendInfo.setPushApp(app); + sendInfo.setPushStream(stream); + sendInfo.setPushSSRC(ssrc); + + Map param; + + + param = new HashMap<>(); + param.put("vhost","__defaultVhost__"); + param.put("app",app); + param.put("stream",stream); + param.put("ssrc", ssrc); + + param.put("dst_url", dstIp); + param.put("dst_port", dstPort); + String is_Udp = isUdp ? "1" : "0"; + param.put("is_udp", is_Udp); + param.put("src_port", sendInfo.getSendLocalPort()); + param.put("use_ps", "0"); + param.put("only_audio", "1"); + + + Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream); + if (streamReady) { + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param); + if (jsonObject.getInteger("code") == 0) { + logger.info("[第三方PS服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, param); + redisTemplate.opsForValue().set(key, sendInfo); + }else { + redisTemplate.delete(key); + logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg")); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg")); + } + }else { + logger.info("[第三方PS服务对接->发送流] 流不存在，等待流上线，callId->{}", callId); + String uuid = UUID.randomUUID().toString(); + HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId()); + dynamicTask.startDelay(uuid, ()->{ + logger.info("[第三方PS服务对接->发送流] 等待流上线超时 callId->{}", callId); + redisTemplate.delete(key); + hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + }, 10000); + + // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 + OtherPsSendInfo finalSendInfo = sendInfo; + hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + hookSubscribe.addSubscribe(hookSubscribeForStreamChange, + (mediaServerItemInUse, response)->{ + dynamicTask.stop(uuid); + logger.info("[第三方PS服务对接->发送流] 流上线，开始发流 callId->{}", callId); + try { + Thread.sleep(400); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param); + if (jsonObject.getInteger("code") == 0) { + logger.info("[第三方PS服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, param); + redisTemplate.opsForValue().set(key, finalSendInfo); + }else { + redisTemplate.delete(key); + logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg")); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg")); + } + hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + }); + } + } + + @GetMapping(value = "/send/stop") + @ResponseBody + @Operation(summary = "关闭发送流") + @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true) + public void closeSendRTP(String callId) { + logger.info("[第三方PS服务对接->关闭发送流] callId->{}", callId); + String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; + OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key); + if (sendInfo == null){ + throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流"); + } + Map param = new HashMap<>(); + param.put("vhost","__defaultVhost__"); + param.put("app",sendInfo.getPushApp()); + param.put("stream",sendInfo.getPushStream()); + param.put("ssrc",sendInfo.getPushSSRC()); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + Boolean result = zlmServerFactory.stopSendRtpStream(mediaServerItem, param); + if (!result) { + logger.info("[第三方PS服务对接->关闭发送流] 失败 callId->{}", callId); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "停止发流失败"); + }else { + logger.info("[第三方PS服务对接->关闭发送流] 成功 callId->{}", callId); + } + redisTemplate.delete(key); + } + + + @GetMapping(value = "/getTestPort") + @ResponseBody + public int getTestPort() { + MediaServerItem defaultMediaServer = mediaServerService.getDefaultMediaServer(); + +// for (int i = 0; i <300; i++) { +// new Thread(() -> { +// int nextPort = sendRtpPortManager.getNextPort(defaultMediaServer); +// try { +// Thread.sleep((int)Math.random()*10); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } +// System.out.println(nextPort); +// }).start(); +// } + + return sendRtpPortManager.getNextPort(defaultMediaServer); + } +}