From 287d42c1968a53c97ee2eff20bed444d842a41b5 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 14 七月 2023 11:27:58 +0800
Subject: [PATCH] 推流默认开启音频
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java | 234 ++++++++++++++++++++++++++++++++--------------------------
1 files changed, 130 insertions(+), 104 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
index 61dea12..33eb5c1 100644
--- 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,18 +2,17 @@
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.ZlmHttpHookSubscribe;
-import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
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.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.*;
-import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.slf4j.Logger;
@@ -22,11 +21,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,7 +69,7 @@
@Autowired
- private ZLMMediaListManager zlmMediaListManager;
+ private RedisTemplate<Object, Object> redisTemplate;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@@ -76,19 +77,13 @@
@Autowired
private IMediaServerService mediaServerService;
- @Autowired
- private IRedisCatchStorage redisCatchStorage;
@Autowired
private DynamicTask dynamicTask;
- @Autowired
- private ZLMMediaListManager mediaListManager;
@Autowired
private ZlmHttpHookSubscribe subscribe;
-
- private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@@ -111,107 +106,106 @@
@Override
public void onMessage(Message message, byte[] bytes) {
-
+ boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message);
- if (!taskQueueHandlerRun) {
- taskQueueHandlerRun = true;
+ if (isEmpty) {
taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
- JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
- WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
- if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
- continue;
- }
- if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
- logger.info("[鏀跺埌REDIS閫氱煡] 璇锋眰锛� {}", new String(msg.getBody()));
-
- switch (wvpRedisMsg.getCmd()){
- case WvpRedisMsgCmd.GET_SEND_ITEM:
- RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
- requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
- break;
- case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
- RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());;
- requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
- break;
- default:
- break;
+ try {
+ JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
+ WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
+ if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
+ continue;
}
+ if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
+ logger.info("[鏀跺埌REDIS閫氱煡] 璇锋眰锛� {}", new String(msg.getBody()));
- }else {
- logger.info("[鏀跺埌REDIS閫氱煡] 鍥炲锛� {}", new String(msg.getBody()));
- switch (wvpRedisMsg.getCmd()){
- case WvpRedisMsgCmd.GET_SEND_ITEM:
+ switch (wvpRedisMsg.getCmd()){
+ case WvpRedisMsgCmd.GET_SEND_ITEM:
+ RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
+ requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
+ break;
+ case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
+ RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
+ requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
- WVPResult content = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
+ break;
+ default:
+ break;
+ }
- String key = wvpRedisMsg.getSerial();
- switch (content.getCode()) {
- case 0:
- ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
- PlayMsgCallback playMsgCallback = callbacks.get(key);
- if (playMsgCallback != null) {
- callbacksForError.remove(key);
- try {
- playMsgCallback.handler(responseSendItemMsg);
- } catch (ParseException e) {
- logger.error("[REDIS娑堟伅澶勭悊寮傚父] ", e);
+ }else {
+ logger.info("[鏀跺埌REDIS閫氱煡] 鍥炲锛� {}", new String(msg.getBody()));
+ switch (wvpRedisMsg.getCmd()){
+ case WvpRedisMsgCmd.GET_SEND_ITEM:
+
+ WVPResult content = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
+
+ String key = wvpRedisMsg.getSerial();
+ switch (content.getCode()) {
+ case 0:
+ ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
+ PlayMsgCallback playMsgCallback = callbacks.get(key);
+ if (playMsgCallback != null) {
+ callbacksForError.remove(key);
+ try {
+ playMsgCallback.handler(responseSendItemMsg);
+ } catch (ParseException e) {
+ logger.error("[REDIS娑堟伅澶勭悊寮傚父] ", e);
+ }
}
- }
- break;
- case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
- case ERROR_CODE_OFFLINE:
- case ERROR_CODE_TIMEOUT:
- PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
- if (errorCallback != null) {
- callbacks.remove(key);
- errorCallback.handler(content);
- }
- break;
- default:
- break;
- }
- break;
- case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
- WVPResult wvpResult = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
- String serial = wvpRedisMsg.getSerial();
- switch (wvpResult.getCode()) {
- case 0:
- JSONObject jsonObject = (JSONObject)wvpResult.getData();
- PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
- if (playMsgCallback != null) {
- callbacksForError.remove(serial);
- playMsgCallback.handler(jsonObject);
- }
- break;
- case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
- case ERROR_CODE_OFFLINE:
- case ERROR_CODE_TIMEOUT:
- PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
- if (errorCallback != null) {
- callbacks.remove(serial);
- errorCallback.handler(wvpResult);
- }
- break;
- default:
- break;
- }
- break;
- default:
- break;
+ break;
+ case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
+ case ERROR_CODE_OFFLINE:
+ case ERROR_CODE_TIMEOUT:
+ PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
+ if (errorCallback != null) {
+ callbacks.remove(key);
+ errorCallback.handler(content);
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
+ WVPResult wvpResult = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
+ String serial = wvpRedisMsg.getSerial();
+ switch (wvpResult.getCode()) {
+ case 0:
+ JSONObject jsonObject = (JSONObject)wvpResult.getData();
+ PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
+ if (playMsgCallback != null) {
+ callbacksForError.remove(serial);
+ playMsgCallback.handler(jsonObject);
+ }
+ break;
+ case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
+ case ERROR_CODE_OFFLINE:
+ case ERROR_CODE_TIMEOUT:
+ PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
+ if (errorCallback != null) {
+ callbacks.remove(serial);
+ errorCallback.handler(wvpResult);
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+
}
+ }catch (Exception e) {
+ logger.warn("[RedisGbPlayMsg] 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
+ logger.error("[RedisGbPlayMsg] 寮傚父鍐呭锛� ", e);
}
}
- taskQueueHandlerRun = false;
});
}
-
-
-
-
-
-
}
/**
@@ -250,7 +244,7 @@
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -269,7 +263,7 @@
WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
return;
}
// 纭畾娴佹槸鍚﹀湪绾�
@@ -292,7 +286,7 @@
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}, userSetting.getPlatformPlayTimeout());
// 娣诲姞璁㈤槄
@@ -306,7 +300,12 @@
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));
+
+// redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
}
}
@@ -331,7 +330,7 @@
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -368,7 +367,7 @@
wvpResult.setMsg("timeout");
errorCallback.handler(wvpResult);
}, userSetting.getPlatformPlayTimeout());
- RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
+ redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -393,6 +392,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