From 090d5b8b624798179482817ae7f29e519d85fa54 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 3 Jul 2023 16:53:03 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E4=B8=8E=E7=AC=AC=E4=B8=89?= =?UTF-8?q?=E6=96=B9=E5=AF=B9=E6=8E=A5=E7=9A=84=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/common/VideoManagerConstants.java | 1 + .../vmp/vmanager/bean/OtherRtpSendInfo.java | 135 ++++++++++++++ .../iot/vmp/vmanager/rtp/RtpController.java | 165 ++++++++++++++++-- 3 files changed, 290 insertions(+), 11 deletions(-) create mode 100644 src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherRtpSendInfo.java 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 8105c0be..6eec845d 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -153,6 +153,7 @@ 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_"; /** * Redis Const diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherRtpSendInfo.java b/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherRtpSendInfo.java new file mode 100644 index 00000000..225e40c4 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherRtpSendInfo.java @@ -0,0 +1,135 @@ +package com.genersoft.iot.vmp.vmanager.bean; + +public class OtherRtpSendInfo { + + /** + * 发流IP + */ + private String ip; + + /** + * 发流端口 + */ + private int port; + + /** + * 收流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 getIp() { + return ip; + } + + public void setIp(String ip) { + this.ip = ip; + } + + public int getPort() { + return port; + } + + public void setPort(int port) { + this.port = port; + } + + 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 "OtherRtpSendInfo{" + + "ip='" + ip + '\'' + + ", port=" + port + + ", receiveIp='" + receiveIp + '\'' + + ", receivePort=" + receivePort + + ", callId='" + callId + '\'' + + ", stream='" + stream + '\'' + + '}'; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java index c8c1625c..5c74fcc9 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java @@ -1,25 +1,42 @@ package com.genersoft.iot.vmp.vmanager.rtp; +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.SipConfig; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.VersionInfo; import com.genersoft.iot.vmp.conf.exception.ControllerException; -import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; 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.MediaServerItem; -import com.genersoft.iot.vmp.service.*; +import com.genersoft.iot.vmp.service.IDeviceChannelService; +import com.genersoft.iot.vmp.service.IDeviceService; +import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo; 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.beans.factory.annotation.Value; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.RestController; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + @SuppressWarnings("rawtypes") @Tag(name = "第三方服务对接") @@ -27,8 +44,13 @@ import org.springframework.web.bind.annotation.RestController; @RequestMapping("/api/rtp") public class RtpController { + private final static Logger logger = LoggerFactory.getLogger(RtpController.class); + @Autowired - private ZlmHttpHookSubscribe zlmHttpHookSubscribe; + private ZLMRTPServerFactory zlmServerFactory; + + @Autowired + private ZlmHttpHookSubscribe hookSubscribe; @Autowired private IMediaServerService mediaServerService; @@ -49,11 +71,11 @@ public class RtpController { private IDeviceChannelService channelService; @Autowired - private IStreamPushService pushService; + private DynamicTask dynamicTask; @Autowired - private IStreamProxyService proxyService; + private RedisTemplate redisTemplate; @Value("${server.port}") @@ -73,12 +95,76 @@ public class RtpController { @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 SendRtpItem openRtpServer(Boolean isSend, String ssrc, String callId, String stream, Integer tcpMode, String callBack) { - MediaServerItem mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null); + public OtherRtpSendInfo openRtpServer(Boolean isSend, String ssrc, String callId, String stream, Integer tcpMode, String callBack) { + + logger.info("[第三方服务对接->开启收流和获取发流信息] 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"); } - return null; + 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格式错误"); + } + + } + int localPort = zlmServerFactory.createRTPServer(mediaServerItem, stream, ssrcInt, null, false, tcpMode); + // 注册回调如果rtp收流超时则通过回调发送通知 + if (callBack != null) { + HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, mediaServerItem.getId()); + // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 + hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, + (mediaServerItemInUse, response)->{ + if (stream.equals(response.getString("stream_id"))) { + logger.info("[开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId); + 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("[开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e); + } + } + }); + } + OtherRtpSendInfo otherRtpSendInfo = new OtherRtpSendInfo(); + otherRtpSendInfo.setReceiveIp(mediaServerItem.getSdpIp()); + otherRtpSendInfo.setReceivePort(localPort); + otherRtpSendInfo.setCallId(callId); + otherRtpSendInfo.setStream(stream); + if (isSend != null && isSend) { + String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId; + // 预创建发流信息 + int port = zlmServerFactory.keepPort(mediaServerItem, callId, 0, ssrc1 -> { + return redisTemplate.opsForValue().get(key) != null; + }); + + // 将信息写入redis中，以备后用 + redisTemplate.opsForValue().set(key, otherRtpSendInfo); + // 设置超时任务，超时未使用，则自动移除，并关闭端口保持, 默认五分钟 + dynamicTask.startDelay(key, ()->{ + logger.info("[第三方服务对接->开启收流和获取发流信息] 端口保持超时 callId->{}", callId); + redisTemplate.delete(key); + zlmServerFactory.releasePort(mediaServerItem, callId); + }, 300000); + otherRtpSendInfo.setIp(mediaServerItem.getSdpIp()); + otherRtpSendInfo.setPort(port); + logger.info("[开启收流和获取发流信息] 结果，callId->{}， {}", callId, otherRtpSendInfo); + } + return otherRtpSendInfo; } @GetMapping(value = "/receive/close") @@ -86,7 +172,9 @@ public class RtpController { @Operation(summary = "关闭收流") @Parameter(name = "stream", description = "流的ID", required = true) public void closeRtpServer(String stream) { - + logger.info("[第三方服务对接->关闭收流] stream->{}", stream); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + zlmServerFactory.closeRtpServer(mediaServerItem,stream); } @GetMapping(value = "/send/start") @@ -99,9 +187,46 @@ public class RtpController { @Parameter(name = "stream", description = "待发送流Id", required = true) @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true) @Parameter(name = "onlyAudio", description = "是否只有音频", required = true) + @Parameter(name = "isUdp", description = "是否为UDP", required = true) @Parameter(name = "streamType", description = "流类型，1为es流，2为ps流， 默认es流", required = false) - public void sendRTP(String ssrc, String ip, Integer port, String app, String stream, String callId, Boolean onlyAudio, Integer streamType) { + public void sendRTP(String ssrc, String ip, Integer port, String app, String stream, String callId, Boolean onlyAudio, Boolean isUdp, Integer streamType) { + logger.info("[第三方服务对接->发送流] ssrc->{}, ip->{}, port->{}, app->{}, stream->{}, callId->{}, onlyAudio->{}, streamType->{}", + ssrc, ip, port, app, stream, callId, onlyAudio, streamType == 1? "ES":"PS"); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId; + OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key); + if (sendInfo != null) { + zlmServerFactory.releasePort(mediaServerItem, sendInfo.getCallId()); + }else { + sendInfo = new OtherRtpSendInfo(); + } + sendInfo.setPushApp(app); + sendInfo.setPushStream(stream); + sendInfo.setPushSSRC(ssrc); + Map param = new HashMap<>(12); + param.put("vhost","__defaultVhost__"); + param.put("app",app); + param.put("stream",stream); + param.put("ssrc", ssrc); + + param.put("dst_url",ip); + param.put("dst_port", port); + String is_Udp = isUdp ? "1" : "0"; + param.put("is_udp", is_Udp); + param.put("src_port", sendInfo.getPort()); + param.put("use_ps", streamType==2 ? "1" : "0"); + param.put("only_audio", onlyAudio ? "1" : "0"); + + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param); + if (jsonObject.getInteger("code") == 0) { + logger.info("[第三方服务对接->发送流] 发流成功，callId->{}", callId); + redisTemplate.opsForValue().set(key, sendInfo); + }else { + redisTemplate.delete(key); + logger.info("[第三方服务对接->发送流] 发流失败，callId->{}, {}", callId, jsonObject.getString("msg")); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "[发流失败] " + jsonObject.getString("msg")); + } } @@ -111,7 +236,25 @@ public class RtpController { @Operation(summary = "关闭发送流") @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true) public void closeSendRTP(String callId) { - + logger.info("[第三方服务对接->关闭发送流] callId->{}", callId); + String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId; + OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)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("[第三方服务对接->关闭发送流] 失败 callId->{}", callId); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "停止发流失败"); + }else { + logger.info("[第三方服务对接->关闭发送流] 成功 callId->{}", callId); + } } }