From b1b6fae22c5ab3013d73df1a43cd42f7fe0fa347 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期二, 09 四月 2024 10:35:11 +0800 Subject: [PATCH] Merge branch 'master' into dev/zlm --- src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java | 100 ++++++++++++++++++++++++------------------------- 1 files changed, 49 insertions(+), 51 deletions(-) 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 old mode 100644 new mode 100755 index ba4473f..7bd2310 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java @@ -5,19 +5,20 @@ 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.conf.security.JwtUtils; +import com.genersoft.iot.vmp.media.event.hook.Hook; +import com.genersoft.iot.vmp.media.event.hook.HookType; import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; -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.HookSubscribeForStreamChange; -import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; +import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; +import com.genersoft.iot.vmp.media.bean.MediaServer; +import com.genersoft.iot.vmp.media.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.OtherRtpSendInfo; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.security.SecurityRequirement; import io.swagger.v3.oas.annotations.tags.Tag; import okhttp3.OkHttpClient; import okhttp3.Request; @@ -42,19 +43,19 @@ @RequestMapping("/api/rtp") public class RtpController { - private final static Logger logger = LoggerFactory.getLogger(RtpController.class); - @Autowired - private ZLMRTPServerFactory zlmServerFactory; - - @Autowired - private ZlmHttpHookSubscribe hookSubscribe; - - @Autowired - private IMediaServerService mediaServerService; + private ZLMServerFactory zlmServerFactory; @Autowired private SendRtpPortManager sendRtpPortManager; + + private final static Logger logger = LoggerFactory.getLogger(RtpController.class); + + @Autowired + private HookSubscribe hookSubscribe; + + @Autowired + private IMediaServerService mediaServerService; @Autowired private UserSetting userSetting; @@ -62,14 +63,13 @@ @Autowired private DynamicTask dynamicTask; - @Autowired private RedisTemplate<Object, Object> redisTemplate; @GetMapping(value = "/receive/open") @ResponseBody - @Operation(summary = "寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅") + @Operation(summary = "寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅", security = @SecurityRequirement(name = JwtUtils.HEADER)) @Parameter(name = "isSend", description = "鏄惁鍙戦�侊紝false鏃跺彧寮�鍚敹娴侊紝 true鍚屾椂杩斿洖鎺ㄦ祦淇℃伅", required = true) @Parameter(name = "callId", description = "鏁翠釜杩囩▼鐨勫敮涓�鏍囪瘑锛屼负浜嗕笌鍚庣画鎺ュ彛鍏宠仈", required = true) @Parameter(name = "ssrc", description = "鏉ユ簮娴佺殑SSRC锛屼笉浼犲垯涓嶆牎楠屾潵婧恠src", required = false) @@ -81,7 +81,7 @@ logger.info("[绗笁鏂规湇鍔″鎺�->寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅] isSend->{}, ssrc->{}, callId->{}, stream->{}, tcpMode->{}, callBack->{}", isSend, ssrc, callId, stream, tcpMode==0?"UDP":"TCP琚姩", callBack); - MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer(); if (mediaServerItem == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(),"娌℃湁鍙敤鐨凪ediaServer"); } @@ -91,30 +91,28 @@ if (isSend != null && isSend && callId == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(),"isSend涓簍rue鏃讹紝CallID涓嶈兘涓虹┖"); } - int ssrcInt = 0; + long ssrcInt = 0; if (ssrc != null) { try { - ssrcInt = Integer.parseInt(ssrc); + ssrcInt = Long.parseLong(ssrc); }catch (NumberFormatException e) { throw new ControllerException(ErrorCode.ERROR100.getCode(),"ssrc鏍煎紡閿欒"); } } String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + callId + "_" + stream; - int localPortForVideo = zlmServerFactory.createRTPServer(mediaServerItem, stream, ssrcInt, null, false, tcpMode); - int localPortForAudio = zlmServerFactory.createRTPServer(mediaServerItem, stream + "_a" , ssrcInt, null, false, tcpMode); + int localPortForVideo = zlmServerFactory.createRTPServer(mediaServerItem, stream, ssrcInt, null, false, false, tcpMode); + int localPortForAudio = zlmServerFactory.createRTPServer(mediaServerItem, stream + "_a" , ssrcInt, null, false, false, tcpMode); if (localPortForVideo == 0 || localPortForAudio == 0) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "鑾峰彇绔彛澶辫触"); } // 娉ㄥ唽鍥炶皟濡傛灉rtp鏀舵祦瓒呮椂鍒欓�氳繃鍥炶皟鍙戦�侀�氱煡 if (callBack != null) { - HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId()); + Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServerItem.getId()); // 璁㈤槄 zlm鍚姩浜嬩欢, 鏂扮殑zlm涔熶細浠庤繖閲岃繘鍏ョ郴缁� - hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, - (mediaServerItemInUse, response)->{ - if (stream.equals(response.getString("stream_id"))) { - logger.info("[绗笁鏂规湇鍔″鎺�->寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅] 绛夊緟鏀舵祦瓒呮椂 callId->{}, 鍙戦�佸洖璋�", callId); - // 灏嗕俊鎭啓鍏edis涓紝浠ュ鍚庣敤 - redisTemplate.delete(receiveKey); + hookSubscribe.addSubscribe(hook, + (hookData)->{ + if (stream.equals(hookData.getStream())) { + logger.info("[寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅] 绛夊緟鏀舵祦瓒呮椂 callId->{}, 鍙戦�佸洖璋�", callId); OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder(); OkHttpClient client = httpClientBuilder.build(); String url = callBack + "?callId=" + callId; @@ -124,10 +122,11 @@ } catch (IOException e) { logger.error("[绗笁鏂规湇鍔″鎺�->寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅] 绛夊緟鏀舵祦瓒呮椂 callId->{}, 鍙戦�佸洖璋冨け璐�", callId, e); } - hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout); + hookSubscribe.removeSubscribe(hook); } }); } + String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_" + callId; OtherRtpSendInfo otherRtpSendInfo = new OtherRtpSendInfo(); otherRtpSendInfo.setReceiveIp(mediaServerItem.getSdpIp()); otherRtpSendInfo.setReceivePortForVideo(localPortForVideo); @@ -138,7 +137,6 @@ // 灏嗕俊鎭啓鍏edis涓紝浠ュ鍚庣敤 redisTemplate.opsForValue().set(receiveKey, otherRtpSendInfo); if (isSend != null && isSend) { - String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_" + callId; // 棰勫垱寤哄彂娴佷俊鎭� int portForVideo = sendRtpPortManager.getNextPort(mediaServerItem); int portForAudio = sendRtpPortManager.getNextPort(mediaServerItem); @@ -150,16 +148,18 @@ redisTemplate.opsForValue().set(key, otherRtpSendInfo, 300, TimeUnit.SECONDS); logger.info("[绗笁鏂规湇鍔″鎺�->寮�鍚敹娴佸拰鑾峰彇鍙戞祦淇℃伅] 缁撴灉锛宑allId->{}锛� {}", callId, otherRtpSendInfo); } + // 灏嗕俊鎭啓鍏edis涓紝浠ュ鍚庣敤 + redisTemplate.opsForValue().set(key, otherRtpSendInfo, 300, TimeUnit.SECONDS); return otherRtpSendInfo; } @GetMapping(value = "/receive/close") @ResponseBody - @Operation(summary = "鍏抽棴鏀舵祦") + @Operation(summary = "鍏抽棴鏀舵祦", security = @SecurityRequirement(name = JwtUtils.HEADER)) @Parameter(name = "stream", description = "娴佺殑ID", required = true) public void closeRtpServer(String stream) { logger.info("[绗笁鏂规湇鍔″鎺�->鍏抽棴鏀舵祦] stream->{}", stream); - MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer(); zlmServerFactory.closeRtpServer(mediaServerItem,stream); zlmServerFactory.closeRtpServer(mediaServerItem,stream + "_a"); String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_*_" + stream; @@ -174,7 +174,7 @@ @GetMapping(value = "/send/start") @ResponseBody - @Operation(summary = "鍙戦�佹祦") + @Operation(summary = "鍙戦�佹祦", security = @SecurityRequirement(name = JwtUtils.HEADER)) @Parameter(name = "ssrc", description = "鍙戦�佹祦鐨凷SRC", required = true) @Parameter(name = "dstIpForAudio", description = "鐩爣闊抽鏀舵祦IP", required = false) @Parameter(name = "dstIpForVideo", description = "鐩爣瑙嗛鏀舵祦IP", required = false) @@ -222,7 +222,7 @@ if (!((dstPortForAudio > 0 && !ObjectUtils.isEmpty(dstPortForAudio) || (dstPortForVideo > 0 && !ObjectUtils.isEmpty(dstIpForVideo))))) { throw new ControllerException(ErrorCode.ERROR400.getCode(), "鑷冲皯搴旇瀛樺湪涓�缁勯煶棰戞垨瑙嗛鍙戦�佸弬鏁�"); } - MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + MediaServer mediaServer = mediaServerService.getDefaultMediaServer(); String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_" + callId; OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key); if (sendInfo == null) { @@ -246,7 +246,6 @@ String is_Udp = isUdp ? "1" : "0"; paramForAudio.put("is_udp", is_Udp); paramForAudio.put("src_port", sendInfo.getSendLocalPortForAudio()); - paramForAudio.put("use_ps", "0"); paramForAudio.put("only_audio", "1"); if (ptForAudio != null) { paramForAudio.put("pt", ptForAudio); @@ -267,7 +266,6 @@ String is_Udp = isUdp ? "1" : "0"; paramForVideo.put("is_udp", is_Udp); paramForVideo.put("src_port", sendInfo.getSendLocalPortForVideo()); - paramForVideo.put("use_ps", "0"); paramForVideo.put("only_audio", "0"); if (ptForVideo != null) { paramForVideo.put("pt", ptForVideo); @@ -277,10 +275,10 @@ paramForVideo = null; } - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream); + Boolean streamReady = zlmServerFactory.isStreamReady(mediaServer, app, stream); if (streamReady) { if (paramForVideo != null) { - JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForVideo); + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForVideo); if (jsonObject.getInteger("code") == 0) { logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 瑙嗛娴佸彂娴佹垚鍔燂紝callId->{}锛宲aram->{}", callId, paramForVideo); redisTemplate.opsForValue().set(key, sendInfo); @@ -291,7 +289,7 @@ } } if(paramForAudio != null) { - JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForAudio); + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForAudio); if (jsonObject.getInteger("code") == 0) { logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 闊抽娴佸彂娴佹垚鍔燂紝callId->{}锛宲aram->{}", callId, paramForAudio); redisTemplate.opsForValue().set(key, sendInfo); @@ -304,18 +302,18 @@ }else { logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 娴佷笉瀛樺湪锛岀瓑寰呮祦涓婄嚎锛宑allId->{}", callId); String uuid = UUID.randomUUID().toString(); - HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId()); + Hook hook = Hook.getInstance(HookType.on_media_arrival, app, stream, mediaServer.getId()); dynamicTask.startDelay(uuid, ()->{ logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 绛夊緟娴佷笂绾胯秴鏃� callId->{}", callId); redisTemplate.delete(key); - hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + hookSubscribe.removeSubscribe(hook); }, 10000); // 璁㈤槄 zlm鍚姩浜嬩欢, 鏂扮殑zlm涔熶細浠庤繖閲岃繘鍏ョ郴缁� OtherRtpSendInfo finalSendInfo = sendInfo; - hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); - hookSubscribe.addSubscribe(hookSubscribeForStreamChange, - (mediaServerItemInUse, response)->{ + hookSubscribe.removeSubscribe(hook); + hookSubscribe.addSubscribe(hook, + (hookData)->{ dynamicTask.stop(uuid); logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 娴佷笂绾匡紝寮�濮嬪彂娴� callId->{}", callId); try { @@ -324,7 +322,7 @@ throw new RuntimeException(e); } if (paramForVideo != null) { - JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForVideo); + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForVideo); if (jsonObject.getInteger("code") == 0) { logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 瑙嗛娴佸彂娴佹垚鍔燂紝callId->{}锛宲aram->{}", callId, paramForVideo); redisTemplate.opsForValue().set(key, finalSendInfo); @@ -335,7 +333,7 @@ } } if(paramForAudio != null) { - JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForAudio); + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForAudio); if (jsonObject.getInteger("code") == 0) { logger.info("[绗笁鏂规湇鍔″鎺�->鍙戦�佹祦] 闊抽娴佸彂娴佹垚鍔燂紝callId->{}锛宲aram->{}", callId, paramForAudio); redisTemplate.opsForValue().set(key, finalSendInfo); @@ -345,14 +343,14 @@ throw new ControllerException(ErrorCode.ERROR100.getCode(), "[闊抽娴佸彂娴佸け璐 " + jsonObject.getString("msg")); } } - hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + hookSubscribe.removeSubscribe(hook); }); } } @GetMapping(value = "/send/stop") @ResponseBody - @Operation(summary = "鍏抽棴鍙戦�佹祦") + @Operation(summary = "鍏抽棴鍙戦�佹祦", security = @SecurityRequirement(name = JwtUtils.HEADER)) @Parameter(name = "callId", description = "鏁翠釜杩囩▼鐨勫敮涓�鏍囪瘑锛屼笉浼犲垯浣跨敤闅忔満绔彛鍙戞祦", required = true) public void closeSendRTP(String callId) { logger.info("[绗笁鏂规湇鍔″鎺�->鍏抽棴鍙戦�佹祦] callId->{}", callId); @@ -366,7 +364,7 @@ param.put("app",sendInfo.getPushApp()); param.put("stream",sendInfo.getPushStream()); param.put("ssrc",sendInfo.getPushSSRC()); - MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer(); Boolean result = zlmServerFactory.stopSendRtpStream(mediaServerItem, param); if (!result) { logger.info("[绗笁鏂规湇鍔″鎺�->鍏抽棴鍙戦�佹祦] 澶辫触 callId->{}", callId); -- Gitblit v1.8.0