From 5e73874880cdfd5b6b99147a0cdd8a6eabcfbf16 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期四, 22 九月 2022 11:22:08 +0800
Subject: [PATCH] 添加队列处理redis消息和sip消息，支持使用推流状态作为通道在线状态
---
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 75 ++++++++++++++++++++++++++++++++-----
1 files changed, 64 insertions(+), 11 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 78b59d9..f645b8a 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;
@@ -105,7 +103,7 @@
@PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
public JSONObject onServerKeepalive(@RequestBody JSONObject json){
- logger.info("[ ZLM HOOK ] on_server_keepalive API璋冪敤锛屽弬鏁帮細" + json.toString());
+ logger.info("[ ZLM HOOK ]on_server_keepalive API璋冪敤锛屽弬鏁帮細" + json.toString());
String mediaServerId = json.getString("mediaServerId");
List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
if (subscribes != null && subscribes.size() > 0) {
@@ -455,6 +453,7 @@
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
// 濡傛灉姝ｅ湪缁欎笂绾ф帹閫侊紝鍒欏彂閫乥ye
+
}else{
streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null);
if (streamInfo != null) {
@@ -511,6 +510,19 @@
}
}
}
+ if (!regist) {
+ List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream);
+ if (sendRtpItems.size() > 0) {
+ for (SendRtpItem sendRtpItem : sendRtpItems) {
+ if (sendRtpItem.getApp().equals(app)) {
+ String platformId = sendRtpItem.getPlatformId();
+ ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
+
+ commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
+ }
+ }
+ }
+ }
}
JSONObject ret = new JSONObject();
@@ -544,6 +556,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());
}
}
}
@@ -573,13 +587,19 @@
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("[{}/{}]<-[{}] 鎷夋祦浠ｇ悊鏃犱汉瑙傜湅宸茬粡绉婚櫎", 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("[{}/{}]<-[{}] 鎷夋祦浠ｇ悊鏃犱汉瑙傜湅宸茬粡绉婚櫎", app, streamId, url);
+ }else if (streamProxyItem.isEnable_disable_none_reader()) {
+ // 鏃犱汉瑙傜湅鍋滅敤
+ ret.put("close", true);
+ }else {
+ ret.put("close", false);
+ }
}
return ret;
}
@@ -626,7 +646,7 @@
@ResponseBody
@PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8")
public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
-
+
if (logger.isDebugEnabled()) {
logger.debug("[ ZLM HOOK ]on_server_started API璋冪敤锛屽弬鏁帮細" + jsonObject.toString());
}
@@ -649,6 +669,39 @@
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) {
HashMap<String, String> map = new HashMap<>();
if (ObjectUtils.isEmpty(params)) {
--
Gitblit v1.8.0