From 2bc284222483d0fb76b861ead4f462421274706d Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 04 五月 2023 15:36:04 +0800 Subject: [PATCH] 优化国标点播下级平台,ssrc更新的时因为旧的端口释放慢导致点播失败的问题,使用新的接口直接更新ssrc --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java | 248 +++++++++++++++++++++++++++++++++---------------- 1 files changed, 168 insertions(+), 80 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java index 210aa72..f3e4d44 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java @@ -1,15 +1,16 @@ package com.genersoft.iot.vmp.media.zlm; -import com.alibaba.fastjson.JSONArray; -import com.alibaba.fastjson.JSONObject; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.CommonCallback; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; -import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.media.zlm.dto.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; import java.util.*; @@ -23,6 +24,9 @@ @Autowired private UserSetting userSetting; + + @Autowired + private ZlmHttpHookSubscribe hookSubscribe; private int[] portRangeArray = new int[2]; @@ -66,7 +70,7 @@ String stream = UUID.randomUUID().toString(); param.put("enable_tcp", 1); param.put("stream_id", stream); - param.put("port", 0); +// param.put("port", 0); JSONObject openRtpServerResultJson = zlmresTfulUtils.openRtpServer(mediaServerItem, param); if (openRtpServerResultJson != null) { @@ -87,22 +91,63 @@ return result; } - public int createRTPServer(MediaServerItem mediaServerItem, String streamId, int ssrc) { + /** + * 寮�鍚痳tpServer + * @param mediaServerItem zlm鏈嶅姟瀹炰緥 + * @param streamId 娴両d + * @param ssrc ssrc + * @param port 绔彛锛� 0/null涓轰娇鐢ㄩ殢鏈� + * @param reUsePort 鏄惁閲嶇敤绔彛 + * @param tcpMode 0/null udp 妯″紡锛�1 tcp 琚姩妯″紡, 2 tcp 涓诲姩妯″紡銆� + * @return + */ + public int createRTPServer(MediaServerItem mediaServerItem, String streamId, int ssrc, Integer port, Boolean reUsePort, Integer tcpMode) { int result = -1; // 鏌ヨ姝tp server 鏄惁宸茬粡瀛樺湪 JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, streamId); - if (rtpInfo != null && rtpInfo.getInteger("code") == 0 && rtpInfo.getBoolean("exist")) { - result = rtpInfo.getInteger("local_port"); + logger.info(JSONObject.toJSONString(rtpInfo)); + if(rtpInfo.getInteger("code") == 0){ + if (rtpInfo.getBoolean("exist")) { + result = rtpInfo.getInteger("local_port"); + if (result == 0) { + // 姝ゆ椂璇存槑rtpServer宸茬粡鍒涘缓浣嗘槸娴佽繕娌℃湁鎺ㄤ笂鏉� + // 姝ゆ椂閲嶆柊鎵撳紑rtpServer + Map<String, Object> param = new HashMap<>(); + param.put("stream_id", streamId); + JSONObject jsonObject = zlmresTfulUtils.closeRtpServer(mediaServerItem, param); + if (jsonObject != null ) { + if (jsonObject.getInteger("code") == 0) { + return createRTPServer(mediaServerItem, streamId, ssrc, port, reUsePort, tcpMode); + }else { + logger.warn("[寮�鍚痳tpServer], 閲嶅惎RtpServer閿欒"); + } + } + } + return result; + } + }else if(rtpInfo.getInteger("code") == -2){ return result; } + Map<String, Object> param = new HashMap<>(); - // 鎺ㄦ祦绔彛璁剧疆0鍒欎娇鐢ㄩ殢鏈虹鍙� - param.put("enable_tcp", 1); + + if (tcpMode == null) { + tcpMode = 0; + } + param.put("tcp_mode", tcpMode); param.put("stream_id", streamId); - param.put("port", 0); + if (reUsePort != null) { + param.put("re_use_port", reUsePort?"1":"0"); + } + // 鎺ㄦ祦绔彛璁剧疆0鍒欎娇鐢ㄩ殢鏈虹鍙� + if (port == null) { + param.put("port", 0); + }else { + param.put("port", port); + } param.put("ssrc", ssrc); JSONObject openRtpServerResultJson = zlmresTfulUtils.openRtpServer(mediaServerItem, param); - + logger.info(JSONObject.toJSONString(openRtpServerResultJson)); if (openRtpServerResultJson != null) { if (openRtpServerResultJson.getInteger("code") == 0) { result= openRtpServerResultJson.getInteger("port"); @@ -116,7 +161,7 @@ return result; } - public boolean closeRTPServer(MediaServerItem serverItem, String streamId) { + public boolean closeRtpServer(MediaServerItem serverItem, String streamId) { boolean result = false; if (serverItem !=null){ Map<String, Object> param = new HashMap<>(); @@ -136,32 +181,31 @@ return result; } -// private int getPortFromportRange(MediaServerItem mediaServerItem) { -// int currentPort = mediaServerItem.getCurrentPort(); -// if (currentPort == 0) { -// String[] portRangeStrArray = mediaServerItem.getSendRtpPortRange().split(","); -// if (portRangeStrArray.length != 2) { -// portRangeArray[0] = 30000; -// portRangeArray[1] = 30500; -// }else { -// portRangeArray[0] = Integer.parseInt(portRangeStrArray[0]); -// portRangeArray[1] = Integer.parseInt(portRangeStrArray[1]); -// } -// } -// -// if (currentPort == 0 || currentPort++ > portRangeArray[1]) { -// currentPort = portRangeArray[0]; -// mediaServerItem.setCurrentPort(currentPort); -// return portRangeArray[0]; -// } else { -// if (currentPort % 2 == 1) { -// currentPort++; -// } -// currentPort++; -// mediaServerItem.setCurrentPort(currentPort); -// return currentPort; -// } -// } + public void closeRtpServer(MediaServerItem serverItem, String streamId, CommonCallback<Boolean> callback) { + if (serverItem == null) { + callback.run(false); + return; + } + Map<String, Object> param = new HashMap<>(); + param.put("stream_id", streamId); + zlmresTfulUtils.closeRtpServer(serverItem, param, jsonObject -> { + if (jsonObject != null ) { + if (jsonObject.getInteger("code") == 0) { + callback.run(jsonObject.getInteger("hit") == 1); + return; + }else { + logger.error("鍏抽棴RTP Server 澶辫触: " + jsonObject.getString("msg")); + } + }else { + // 妫�鏌LM鐘舵�� + logger.error("鍏抽棴RTP Server 澶辫触: 璇锋鏌LM鏈嶅姟"); + } + callback.run(false); + }); + + + } + /** * 鍒涘缓涓�涓浗鏍囨帹娴� @@ -173,23 +217,17 @@ * @param tcp 鏄惁涓簍cp * @return SendRtpItem */ - public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String deviceId, String channelId, boolean tcp){ + public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String deviceId, String channelId, boolean tcp, boolean rtcp){ - // 浣跨敤RTPServer 鍔熻兘鎵句竴涓彲鐢ㄧ殑绔彛 - String sendRtpPortRange = serverItem.getSendRtpPortRange(); - if (StringUtils.isEmpty(sendRtpPortRange)) { - return null; - } - String[] portRangeStrArray = serverItem.getSendRtpPortRange().split(","); - int localPort = -1; - if (portRangeStrArray.length != 2) { - localPort = getFreePort(serverItem, 30000, 30500, null); - }else { - localPort = getFreePort(serverItem, Integer.parseInt(portRangeStrArray[0]), Integer.parseInt(portRangeStrArray[1]), null); - } - if (localPort == -1) { - logger.error("娌℃湁鍙敤鐨勭鍙�"); - return null; + // 榛樿涓洪殢鏈虹鍙� + int localPort = 0; + if (userSetting.getGbSendStreamStrict()) { + if (userSetting.getGbSendStreamStrict()) { + localPort = keepPort(serverItem, ssrc); + if (localPort == 0) { + return null; + } + } } SendRtpItem sendRtpItem = new SendRtpItem(); sendRtpItem.setIp(ip); @@ -199,6 +237,7 @@ sendRtpItem.setDeviceId(deviceId); sendRtpItem.setChannelId(channelId); sendRtpItem.setTcp(tcp); + sendRtpItem.setRtcp(rtcp); sendRtpItem.setApp("rtp"); sendRtpItem.setLocalPort(localPort); sendRtpItem.setServerId(userSetting.getServerId()); @@ -216,22 +255,14 @@ * @param tcp 鏄惁涓簍cp * @return SendRtpItem */ - public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String app, String stream, String channelId, boolean tcp){ - // 浣跨敤RTPServer 鍔熻兘鎵句竴涓彲鐢ㄧ殑绔彛 - String sendRtpPortRange = serverItem.getSendRtpPortRange(); - if (StringUtils.isEmpty(sendRtpPortRange)) { - return null; - } - String[] portRangeStrArray = serverItem.getSendRtpPortRange().split(","); - int localPort = -1; - if (portRangeStrArray.length != 2) { - localPort = getFreePort(serverItem, 30000, 30500, null); - }else { - localPort = getFreePort(serverItem, Integer.parseInt(portRangeStrArray[0]), Integer.parseInt(portRangeStrArray[1]), null); - } - if (localPort == -1) { - logger.error("娌℃湁鍙敤鐨勭鍙�"); - return null; + public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String app, String stream, String channelId, boolean tcp, boolean rtcp){ + // 榛樿涓洪殢鏈虹鍙� + int localPort = 0; + if (userSetting.getGbSendStreamStrict()) { + localPort = keepPort(serverItem, ssrc); + if (localPort == 0) { + return null; + } } SendRtpItem sendRtpItem = new SendRtpItem(); sendRtpItem.setIp(ip); @@ -245,7 +276,46 @@ sendRtpItem.setLocalPort(localPort); sendRtpItem.setServerId(userSetting.getServerId()); sendRtpItem.setMediaServerId(serverItem.getId()); + sendRtpItem.setRtcp(rtcp); return sendRtpItem; + } + + /** + * 淇濇寔绔彛锛岀洿鍒伴渶瑕侀渶瑕佸彂娴佹椂鍐嶉噴鏀� + */ + public int keepPort(MediaServerItem serverItem, String ssrc) { + int localPort = 0; + Map<String, Object> param = new HashMap<>(3); + param.put("port", 0); + param.put("enable_tcp", 1); + param.put("stream_id", ssrc); + JSONObject jsonObject = zlmresTfulUtils.openRtpServer(serverItem, param); + if (jsonObject.getInteger("code") == 0) { + localPort = jsonObject.getInteger("port"); + HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId()); + // 璁㈤槄 zlm鍚姩浜嬩欢, 鏂扮殑zlm涔熶細浠庤繖閲岃繘鍏ョ郴缁� + hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, + (MediaServerItem mediaServerItem, JSONObject response)->{ + logger.info("[涓婄骇鐐规挱] {}->鐩戝惉绔彛鍒版湡缁х画淇濇寔鐩戝惉", ssrc); + keepPort(serverItem, ssrc); + }); + logger.info("[涓婄骇鐐规挱] {}->鐩戝惉绔彛: {}", ssrc, localPort); + }else { + logger.info("[涓婄骇鐐规挱] 鐩戝惉绔彛澶辫触: {}", ssrc); + } + return localPort; + } + + /** + * 閲婃斁淇濇寔鐨勭鍙� + */ + public boolean releasePort(MediaServerItem serverItem, String ssrc) { + logger.info("[涓婄骇鐐规挱] {}->閲婃斁鐩戝惉绔彛", ssrc); + boolean closeRTPServerResult = closeRtpServer(serverItem, ssrc); + HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId()); + // 璁㈤槄 zlm鍚姩浜嬩欢, 鏂扮殑zlm涔熶細浠庤繖閲岃繘鍏ョ郴缁� + hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout); + return closeRTPServerResult; } /** @@ -259,7 +329,7 @@ * 鏌ヨ寰呰浆鎺ㄧ殑娴佹槸鍚﹀氨缁� */ public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) { - JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem,"rtp", "rtmp", streamId); + JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem,"rtp", "rtsp", streamId); return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")); } @@ -267,8 +337,11 @@ * 鏌ヨ寰呰浆鎺ㄧ殑娴佹槸鍚﹀氨缁� */ public Boolean isStreamReady(MediaServerItem mediaServerItem, String app, String streamId) { - JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtmp", streamId); - return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")); + JSONObject mediaInfo = zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId); + return mediaInfo != null && (mediaInfo.getInteger("code") == 0 + + && mediaInfo.getJSONArray("data") != null + && mediaInfo.getJSONArray("data").size() > 0); } /** @@ -277,16 +350,16 @@ * @return */ public int totalReaderCount(MediaServerItem mediaServerItem, String app, String streamId) { - JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtmp", streamId); - Integer code = mediaInfo.getInteger("code"); + JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", streamId); if (mediaInfo == null) { return 0; } + Integer code = mediaInfo.getInteger("code"); if ( code < 0) { logger.warn("鏌ヨ娴�({}/{})鏄惁鏈夊叾瀹冭鐪嬭�呮椂寰楀埌锛� {}", app, streamId, mediaInfo.getString("msg")); return -1; } - if ( code == 0 && ! mediaInfo.getBoolean("online")) { + if ( code == 0 && mediaInfo.getBoolean("online") != null && !mediaInfo.getBoolean("online")) { logger.warn("鏌ヨ娴�({}/{})鏄惁鏈夊叾瀹冭鐪嬭�呮椂寰楀埌锛� {}", app, streamId, mediaInfo.getString("msg")); return -1; } @@ -300,12 +373,12 @@ Boolean result = false; JSONObject jsonObject = zlmresTfulUtils.stopSendRtp(mediaServerItem, param); if (jsonObject == null) { - logger.error("鍋滄RTP鎺ㄦ祦澶辫触: 璇锋鏌LM鏈嶅姟"); + logger.error("[鍋滄RTP鎺ㄦ祦] 澶辫触: 璇锋鏌LM鏈嶅姟"); } else if (jsonObject.getInteger("code") == 0) { result= true; - logger.info("鍋滄RTP鎺ㄦ祦鎴愬姛"); + logger.info("[鍋滄RTP鎺ㄦ祦] 鎴愬姛"); } else { - logger.error("鍋滄RTP鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"),JSONObject.toJSON(param)); + logger.error("[鍋滄RTP鎺ㄦ祦] 澶辫触: {}, 鍙傛暟锛歿}->\r\n{}",jsonObject.getString("msg"), JSON.toJSON(param), jsonObject); } return result; } @@ -313,4 +386,19 @@ public void closeAllSendRtpStream() { } + + public Boolean updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) { + boolean result = false; + JSONObject jsonObject = zlmresTfulUtils.updateRtpServerSSRC(mediaServerItem, streamId, ssrc); + if (jsonObject == null) { + logger.error("[鏇存柊RTPServer] 澶辫触: 璇锋鏌LM鏈嶅姟"); + } else if (jsonObject.getInteger("code") == 0) { + result= true; + logger.info("[鏇存柊RTPServer] 鎴愬姛"); + } else { + logger.error("[鏇存柊RTPServer] 澶辫触: {}, streamId锛歿}锛宻src锛歿}->\r\n{}",jsonObject.getString("msg"), + streamId, ssrc, jsonObject); + } + return result; + } } -- Gitblit v1.8.0