From 221f99c764b51c3cd284c6e5a41492c26c5c7ffc Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 09 九月 2022 17:06:19 +0800
Subject: [PATCH] 优化sip消息

---
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java |  139 +++++++++++++++++++++++++++++-----------------
 1 files changed, 87 insertions(+), 52 deletions(-)

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 ce46393..b789974 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
@@ -19,8 +19,6 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -103,7 +101,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){
+	public JSONObject onServerKeepalive(@RequestBody JSONObject json){
 
 		logger.info("[ ZLM HOOK ] on_server_keepalive API璋冪敤锛屽弬鏁帮細" + json.toString());
 		String mediaServerId = json.getString("mediaServerId");
@@ -118,7 +116,8 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+
+		return ret;
 	}
 
 	/**
@@ -127,16 +126,15 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_flow_report", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onFlowReport(@RequestBody JSONObject json){
+	public JSONObject onFlowReport(@RequestBody JSONObject json){
 		
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_flow_report API璋冪敤锛屽弬鏁帮細" + json.toString());
 		}
-		String mediaServerId = json.getString("mediaServerId");
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -145,7 +143,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_http_access", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onHttpAccess(@RequestBody JSONObject json){
+	public JSONObject onHttpAccess(@RequestBody JSONObject json){
 		
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_http_access API 璋冪敤锛屽弬鏁帮細" + json.toString());
@@ -156,7 +154,7 @@
 		ret.put("err", "");
 		ret.put("path", "");
 		ret.put("second", 600);
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -165,7 +163,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_play", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onPlay(@RequestBody OnPlayHookParam param){
+	public JSONObject onPlay(@RequestBody OnPlayHookParam param){
 
 		JSONObject json = (JSONObject)JSON.toJSON(param);
 
@@ -187,13 +185,13 @@
 			if (streamAuthorityInfo != null && streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(paramMap.get("callId"))) {
 				ret.put("code", 401);
 				ret.put("msg", "Unauthorized");
-				return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
+				return ret;
 			}
 		}
 
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -202,7 +200,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onPublish(@RequestBody OnPublishHookParam param) {
+	public JSONObject onPublish(@RequestBody OnPublishHookParam param) {
 
 		JSONObject json = (JSONObject) JSON.toJSON(param);
 
@@ -216,7 +214,7 @@
 				logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯涓嶈鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)");
 				ret.put("code", 401);
 				ret.put("msg", "Unauthorized");
-				return new ResponseEntity<>(ret.toString(), HttpStatus.OK);
+				return ret;
 			}
 			Map<String, String> paramMap = urlParamToMap(param.getParams());
 			String sign = paramMap.get("sign");
@@ -224,7 +222,7 @@
 				logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯涓嶈鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)");
 				ret.put("code", 401);
 				ret.put("msg", "Unauthorized");
-				return new ResponseEntity<>(ret.toString(), HttpStatus.OK);
+				return ret;
 			}
 			// 鎺ㄦ祦鑷畾涔夋挱鏀鹃壌鏉冪爜
 			String callId = paramMap.get("callId");
@@ -234,7 +232,7 @@
 				logger.info("鎺ㄦ祦閴存潈澶辫触锛� sign 鏃犳潈闄�: callId={}. sign={}", callId, sign);
 				ret.put("code", 401);
 				ret.put("msg", "Unauthorized");
-				return new ResponseEntity<>(ret.toString(), HttpStatus.OK);
+				return ret;
 			}
 			StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
 			streamAuthorityInfo.setCallId(callId);
@@ -242,12 +240,11 @@
 			// 閴存潈閫氳繃
 			redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
 			// 閫氱煡assist鏂扮殑callId
-			taskExecutor.execute(()->{
-				if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {
+			if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {
+				taskExecutor.execute(()->{
 					assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);
-				}
-			});
-
+				});
+			}
 		}else {
 			zlmMediaListManager.sendStreamEvent(param.getApp(),param.getStream(), param.getMediaServerId());
 		}
@@ -291,10 +288,7 @@
 
 			}
 		}
-
-
-
-		return new ResponseEntity<String>(ret.toString(), HttpStatus.OK);
+		return ret;
 	}
 
 
@@ -305,7 +299,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_record_mp4", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onRecordMp4(@RequestBody JSONObject json){
+	public JSONObject onRecordMp4(@RequestBody JSONObject json){
 		
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_record_mp4 API璋冪敤锛屽弬鏁帮細" + json.toString());
@@ -314,7 +308,7 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	/**
 	 * 褰曞埗hls瀹屾垚鍚庨�氱煡浜嬩欢锛涙浜嬩欢瀵瑰洖澶嶄笉鏁忔劅銆�
@@ -322,7 +316,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_record_ts", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onRecordTs(@RequestBody JSONObject json){
+	public JSONObject onRecordTs(@RequestBody JSONObject json){
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_record_ts API璋冪敤锛屽弬鏁帮細" + json.toString());
@@ -331,7 +325,7 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -340,7 +334,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_rtsp_realm", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onRtspRealm(@RequestBody JSONObject json){
+	public JSONObject onRtspRealm(@RequestBody JSONObject json){
 		
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_rtsp_realm API璋冪敤锛屽弬鏁帮細" + json.toString());
@@ -349,7 +343,7 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("realm", "");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	
@@ -359,7 +353,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_rtsp_auth", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onRtspAuth(@RequestBody JSONObject json){
+	public JSONObject onRtspAuth(@RequestBody JSONObject json){
 		
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_rtsp_auth API璋冪敤锛屽弬鏁帮細" + json.toString());
@@ -369,7 +363,7 @@
 		ret.put("code", 0);
 		ret.put("encrypted", false);
 		ret.put("passwd", "test");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -378,7 +372,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_shell_login", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onShellLogin(@RequestBody JSONObject json){
+	public JSONObject onShellLogin(@RequestBody JSONObject json){
 		
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_shell_login API璋冪敤锛屽弬鏁帮細" + json.toString());
@@ -396,7 +390,7 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -405,7 +399,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_stream_changed", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onStreamChanged(@RequestBody MediaItem item){
+	public JSONObject onStreamChanged(@RequestBody MediaItem item){
 
 		logger.info("[ ZLM HOOK ]on_stream_changed API璋冪敤锛屽弬鏁帮細" + JSONObject.toJSONString(item));
 		String mediaServerId = item.getMediaServerId();
@@ -520,7 +514,7 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -529,7 +523,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onStreamNoneReader(@RequestBody JSONObject json){
+	public JSONObject onStreamNoneReader(@RequestBody JSONObject json){
 
 		logger.info("[ ZLM HOOK ]on_stream_none_reader API璋冪敤锛屽弬鏁帮細" + json.toString());
 		String mediaServerId = json.getString("mediaServerId");
@@ -548,6 +542,8 @@
 						for (SendRtpItem sendRtpItem : sendRtpItems) {
 							ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
 							commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+							redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
+									sendRtpItem.getCallId(), sendRtpItem.getStreamId());
 						}
 					}
 				}
@@ -574,18 +570,24 @@
 			if (mediaServerItem != null && mediaServerItem.getStreamNoneReaderDelayMS() == -1) {
 				ret.put("close", false);
 			}
-			return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+			return ret;
 		}else {
 			StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId);
-			if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) {
-				ret.put("close", true);
-				streamProxyService.del(app, streamId);
-				String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();
-				logger.info("[{}/{}]<-[{}] 鎷夋祦浠g悊鏃犱汉瑙傜湅宸茬粡绉婚櫎",  app, streamId, url);
-			}else {
-				ret.put("close", false);
+			if (streamProxyItem != null ) {
+				if (streamProxyItem.isEnable_remove_none_reader()) {
+					// 鏃犱汉瑙傜湅鑷姩绉婚櫎
+					ret.put("close", true);
+					streamProxyService.del(app, streamId);
+					String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();
+					logger.info("[{}/{}]<-[{}] 鎷夋祦浠g悊鏃犱汉瑙傜湅宸茬粡绉婚櫎",  app, streamId, url);
+				}else if (streamProxyItem.isEnable_disable_none_reader()) {
+					// 鏃犱汉瑙傜湅鍋滅敤
+					ret.put("close", true);
+				}else {
+					ret.put("close", false);
+				}
 			}
-			return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+			return ret;
 		}
 	}
 	
@@ -595,7 +597,7 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onStreamNotFound(@RequestBody JSONObject json){
+	public JSONObject onStreamNotFound(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_stream_not_found API璋冪敤锛屽弬鏁帮細" + json.toString());
 		}
@@ -620,7 +622,7 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
+		return ret;
 	}
 	
 	/**
@@ -629,8 +631,8 @@
 	 */
 	@ResponseBody
 	@PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8")
-	public ResponseEntity<String> onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
-		
+	public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
+
 		if (logger.isDebugEnabled()) {
 			logger.debug("[ ZLM HOOK ]on_server_started API璋冪敤锛屽弬鏁帮細" + jsonObject.toString());
 		}
@@ -650,7 +652,40 @@
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
-		return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
+		return ret;
+	}
+
+	/**
+	 * 鍙戦�乺tp(startSendRtp)琚姩鍏抽棴鏃跺洖璋�
+	 */
+	@ResponseBody
+	@PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8")
+	public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody JSONObject jsonObject){
+
+		logger.info("[ ZLM HOOK ]on_send_rtp_stopped API璋冪敤锛屽弬鏁帮細" + jsonObject);
+
+		JSONObject ret = new JSONObject();
+		ret.put("code", 0);
+		ret.put("msg", "success");
+
+		// 鏌ユ壘瀵瑰簲鐨勪笂绾ф帹娴侊紝鍙戦�佸仠姝�
+		String app = jsonObject.getString("app");
+		if (!"rtp".equals(app)) {
+			return ret;
+		}
+		String stream = jsonObject.getString("stream");
+		List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream);
+		if (sendRtpItems.size() > 0) {
+			for (SendRtpItem sendRtpItem : sendRtpItems) {
+				ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+				commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+				redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
+						sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+			}
+		}
+
+
+		return ret;
 	}
 
 	private Map<String, String> urlParamToMap(String params) {

--
Gitblit v1.8.0