From e1d476a54a7aa72f429e41e5c2957b77edbeb0a7 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 13 三月 2024 15:24:09 +0800
Subject: [PATCH] 处理收到redis推动的推流设备信息内容重复的问题
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java | 98 +++++++++++++++++++++++++++++++++----------------
1 files changed, 66 insertions(+), 32 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
old mode 100644
new mode 100755
index 42cbec5..eb261e3
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
@@ -2,12 +2,12 @@
import com.alibaba.fastjson2.JSON;
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.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
+import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
-import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
-import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@@ -22,11 +22,13 @@
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import java.text.ParseException;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -68,10 +70,10 @@
@Autowired
- private ZLMMediaListManager zlmMediaListManager;
+ private RedisTemplate<Object, Object> redisTemplate;
@Autowired
- private ZLMRTPServerFactory zlmrtpServerFactory;
+ private ZLMServerFactory zlmServerFactory;
@Autowired
private IMediaServerService mediaServerService;
@@ -79,11 +81,10 @@
@Autowired
private IRedisCatchStorage redisCatchStorage;
+
@Autowired
private DynamicTask dynamicTask;
- @Autowired
- private ZLMMediaListManager mediaListManager;
@Autowired
private ZlmHttpHookSubscribe subscribe;
@@ -116,8 +117,8 @@
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
try {
- JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
- WvpRedisMsg wvpRedisMsg = JSON.toJavaObject(msgJSON, WvpRedisMsg.class);
+ WvpRedisMsg wvpRedisMsg = JSON.parseObject(msg.getBody(), WvpRedisMsg.class);
+ logger.info("[鏀跺埌REDIS閫氱煡] 娑堟伅锛� {}", JSON.toJSONString(wvpRedisMsg));
if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
continue;
}
@@ -126,12 +127,13 @@
switch (wvpRedisMsg.getCmd()){
case WvpRedisMsgCmd.GET_SEND_ITEM:
- RequestSendItemMsg content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestSendItemMsg.class);
+ RequestSendItemMsg content = JSON.parseObject(wvpRedisMsg.getContent(), RequestSendItemMsg.class);
requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break;
case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
- RequestPushStreamMsg param = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestPushStreamMsg.class);;
+ RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
+
break;
default:
break;
@@ -142,12 +144,12 @@
switch (wvpRedisMsg.getCmd()){
case WvpRedisMsgCmd.GET_SEND_ITEM:
- WVPResult content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class);
+ WVPResult content = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
String key = wvpRedisMsg.getSerial();
switch (content.getCode()) {
case 0:
- ResponseSendItemMsg responseSendItemMsg =JSON.toJavaObject((JSONObject)content.getData(), ResponseSendItemMsg.class);
+ ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
PlayMsgCallback playMsgCallback = callbacks.get(key);
if (playMsgCallback != null) {
callbacksForError.remove(key);
@@ -172,7 +174,7 @@
}
break;
case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
- WVPResult wvpResult = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class);
+ WVPResult wvpResult = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
String serial = wvpRedisMsg.getSerial();
switch (wvpResult.getCode()) {
case 0:
@@ -199,9 +201,11 @@
default:
break;
}
+
}
}catch (Exception e) {
- logger.warn("[RedisGbPlayMsg] 鍙戠幇鏈鐞嗙殑寮傚父, {}",e.getMessage());
+ logger.warn("[RedisGbPlayMsg] 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
+ logger.error("[RedisGbPlayMsg] 寮傚父鍐呭锛� ", e);
}
}
});
@@ -230,7 +234,7 @@
param.put("pt", requestPushStreamMsg.getPt());
param.put("use_ps", requestPushStreamMsg.isPs() ? "1" : "0");
param.put("only_audio", requestPushStreamMsg.isOnlyAudio() ? "1" : "0");
- JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+ JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaInfo, param);
// 鍥炲娑堟伅
responsePushStream(jsonObject, fromId, serial);
}
@@ -242,9 +246,9 @@
result.setData(content);
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
- WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
+ WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, JSON.toJSONString(result));
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -260,15 +264,15 @@
result.setMsg("娴佸獟浣撲笉瀛樺湪");
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
- WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);
+ WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result));
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
return;
}
// 纭畾娴佹槸鍚﹀湪绾�
- boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
- if (streamReady) {
+ Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
+ if (streamReady != null && streamReady) {
logger.info("[鍥炲鎺ㄦ祦淇℃伅] {}/{}", content.getApp(), content.getStream());
responseSendItem(mediaServerItem, content, toId, serial);
}else {
@@ -283,16 +287,16 @@
WVPResult<SendRtpItem> result = new WVPResult<>();
result.setCode(ERROR_CODE_TIMEOUT);
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
- userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
+ userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}, userSetting.getPlatformPlayTimeout());
// 娣诲姞璁㈤槄
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtsp", mediaServerItem.getId());
- subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
+ subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam)->{
dynamicTask.stop(taskKey);
responseSendItem(mediaServerItem, content, toId, serial);
});
@@ -300,8 +304,10 @@
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, content.getApp(), content.getStream(),
content.getChannelId(), content.getPlatformId(), content.getPlatformName(), content.getServerId(),
content.getMediaServerId());
- redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
+ String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED;
+ logger.info("[redis鍙戦�侀�氱煡] 鎺ㄦ祦琚姹� {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream());
+ redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel));
}
}
@@ -309,7 +315,7 @@
* 灏嗚幏鍙栧埌鐨剆endItem鍙戦�佸嚭鍘�
*/
private void responseSendItem(MediaServerItem mediaServerItem, RequestSendItemMsg content, String toId, String serial) {
- SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
+ SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
content.getPort(), content.getSsrc(), content.getPlatformId(),
content.getApp(), content.getStream(), content.getChannelId(),
content.getTcp(), content.getRtcp());
@@ -320,12 +326,13 @@
responseSendItemMsg.setSendRtpItem(sendRtpItem);
responseSendItemMsg.setMediaServerItem(mediaServerItem);
result.setData(responseSendItemMsg);
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
- userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
+ userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -348,7 +355,7 @@
requestSendItemMsg.setServerId(serverId);
String key = UUID.randomUUID().toString();
WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId, WvpRedisMsgCmd.GET_SEND_ITEM,
- key, requestSendItemMsg);
+ key, JSON.toJSONString(requestSendItemMsg));
JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
logger.info("[璇锋眰鎺ㄦ祦SendItem] {}: {}", serverId, jsonObject);
@@ -362,7 +369,7 @@
wvpResult.setMsg("timeout");
errorCallback.handler(wvpResult);
}, userSetting.getPlatformPlayTimeout());
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -373,7 +380,7 @@
public void sendMsgForStartSendRtpStream(String serverId, RequestPushStreamMsg param, PlayMsgCallbackForStartSendRtpStream callback) {
String key = UUID.randomUUID().toString();
WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
- WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, param);
+ WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, JSON.toJSONString(param));
JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
logger.info("[REDIS 璇锋眰鍏朵粬骞冲彴鎺ㄦ祦] {}: {}", serverId, jsonObject);
@@ -387,6 +394,33 @@
callbacksForStartSendRtpStream.remove(key);
callbacksForError.remove(key);
});
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ }
+
+ private SendRtpItem querySendRTPServer(String platformGbId, String channelId, String streamId, String callId) {
+ if (platformGbId == null) {
+ platformGbId = "*";
+ }
+ if (channelId == null) {
+ channelId = "*";
+ }
+ if (streamId == null) {
+ streamId = "*";
+ }
+ if (callId == null) {
+ callId = "*";
+ }
+ String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX
+ + userSetting.getServerId() + "_*_"
+ + platformGbId + "_"
+ + channelId + "_"
+ + streamId + "_"
+ + callId;
+ List<Object> scan = RedisUtil.scan(redisTemplate, key);
+ if (scan.size() > 0) {
+ return (SendRtpItem)redisTemplate.opsForValue().get(scan.get(0));
+ }else {
+ return null;
+ }
}
}
--
Gitblit v1.8.0