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/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index a90670d8..130f49de 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/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 397a4a0f..d2d9657b 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/media/zlm/SendRtpPortManager.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java index f960c7dc..3f28d02a 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 385255ad..42cbd82b 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.apache.commons.lang3.math.NumberUtils; @@ -194,7 +195,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()) { // 推流鉴权 @@ -246,11 +250,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(); @@ -287,7 +301,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); } } @@ -564,7 +581,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; } @@ -593,7 +610,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 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/InviteStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java index 6e460823..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,10 +94,11 @@ 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); @@ -108,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; @@ -133,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) { @@ -174,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; } @@ -191,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; @@ -222,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/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index cf8bdd24..f2653f70 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 @@ -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); @@ -457,16 +454,13 @@ public class PlayServiceImpl implements IPlayService { logger.warn("[录像回放] 单端口收流时不支持TCP主动方式收流 deviceId: {},channelId:{}", deviceId, channelId); throw new ControllerException(ErrorCode.ERROR100.getCode(), "单端口收流时不支持TCP主动方式收流"); } - String stream = null; - if (newMediaServerItem.isRtpEnable()) { - String startTimeStr = startTime.replace("-", "") - .replace(":", "") - .replace(" ", ""); - String endTimeTimeStr = endTime.replace("-", "") - .replace(":", "") - .replace(" ", ""); - stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr; - } + 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); } @@ -628,44 +622,13 @@ public class PlayServiceImpl implements IPlayService { if (ssrcInResponse != null) { // 单端口 // 重新订阅流上线 - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", - ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); - subscribe.removeSubscribe(hookSubscribe); SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(inviteInfo.getDeviceId(), inviteInfo.getChannelId(), null, inviteInfo.getStream()); streamSession.remove(inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); - - String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase(); - hookSubscribe.getContent().put("stream", stream); - - inviteStreamService.updateInviteInfoForStream(inviteInfo, stream); + inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse); streamSession.put(device.getDeviceId(), channelId, ssrcTransaction.getCallId(), - stream, ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType); - subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> { - logger.info("[Invite 200OK] ssrc修正后收到订阅消息： " + hookParam); - dynamicTask.stop(timeOutTaskKey); - subscribe.removeSubscribe(hookSubscribe); - // 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, 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, device.getDeviceId(), channelId, null, - InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), - streamInfo); - if (inviteSessionType == InviteSessionType.PLAY) { - snapOnPlay(mediaServerItemInUse, device.getDeviceId(), channelId, stream); - } - }); + inviteInfo.getStream(), ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType); } } } 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 00000000..ac984092 --- /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/ps/PsController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java new file mode 100644 index 00000000..81ccb573 --- /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); + } +}