From 29f7a6b6eba350f2c49b744f110d1aa033f77d02 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 10 一月 2024 18:55:14 +0800
Subject: [PATCH] 修复多wvp模式推流时信息存储错误
---
src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java | 10 +++++-----
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java | 9 +++++----
数据库/更新-postgresql-kingbase-2.6.9.sql | 3 +++
数据库/初始化-mysql-2.6.9.sql | 1 +
数据库/初始化-postgresql-kingbase-2.6.9.sql | 1 +
数据库/更新-mysql-2.6.9.sql | 3 +++
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java | 18 +++++++++---------
7 files changed, 27 insertions(+), 18 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java b/src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java
index 12d79cb..74890be 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java
@@ -29,12 +29,12 @@
* 娑堟伅鐨処D
*/
private String serial;
- private Object content;
+ private String content;
private final static String requestTag = "req";
private final static String responseTag = "res";
- public static WvpRedisMsg getRequestInstance(String fromId, String toId, String cmd, String serial, Object content) {
+ public static WvpRedisMsg getRequestInstance(String fromId, String toId, String cmd, String serial, String content) {
WvpRedisMsg wvpRedisMsg = new WvpRedisMsg();
wvpRedisMsg.setType(requestTag);
wvpRedisMsg.setFromId(fromId);
@@ -51,7 +51,7 @@
return wvpRedisMsg;
}
- public static WvpRedisMsg getResponseInstance(String fromId, String toId, String cmd, String serial, Object content) {
+ public static WvpRedisMsg getResponseInstance(String fromId, String toId, String cmd, String serial, String content) {
WvpRedisMsg wvpRedisMsg = new WvpRedisMsg();
wvpRedisMsg.setType(responseTag);
wvpRedisMsg.setFromId(fromId);
@@ -106,11 +106,11 @@
this.cmd = cmd;
}
- public Object getContent() {
+ public String getContent() {
return content;
}
- public void setContent(Object content) {
+ public void setContent(String content) {
this.content = content;
}
}
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 2b0f366..3196613 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
@@ -113,8 +113,8 @@
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
try {
- JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
- WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
+ WvpRedisMsg wvpRedisMsg = JSON.parseObject(msg.getBody(), WvpRedisMsg.class);
+ logger.info("[鏀跺埌REDIS閫氱煡] 娑堟伅锛� {}", JSON.toJSONString(wvpRedisMsg));
if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
continue;
}
@@ -123,7 +123,7 @@
switch (wvpRedisMsg.getCmd()){
case WvpRedisMsgCmd.GET_SEND_ITEM:
- RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
+ RequestSendItemMsg content = JSON.parseObject(wvpRedisMsg.getContent(), RequestSendItemMsg.class);
requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break;
case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
@@ -242,7 +242,7 @@
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);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
@@ -260,7 +260,7 @@
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);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -283,7 +283,7 @@
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);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -324,7 +324,7 @@
result.setData(responseSendItemMsg);
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);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -350,7 +350,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);
@@ -375,7 +375,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);
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
index be0924a..682f07c 100755
--- a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
@@ -13,9 +13,9 @@
public interface StreamPushMapper {
@Insert("INSERT INTO wvp_stream_push (app, stream, total_reader_count, origin_type, origin_type_str, " +
- "push_time, alive_second, media_server_id, update_time, create_time, push_ing, self) VALUES" +
+ "push_time, alive_second, media_server_id, server_id, update_time, create_time, push_ing, self) VALUES" +
"(#{app}, #{stream}, #{totalReaderCount}, #{originType}, #{originTypeStr}, " +
- "#{pushTime}, #{aliveSecond}, #{mediaServerId} , #{updateTime} , #{createTime}, " +
+ "#{pushTime}, #{aliveSecond}, #{mediaServerId} , #{serverId} , #{updateTime} , #{createTime}, " +
"#{pushIng}, #{self} )")
int add(StreamPushItem streamPushItem);
@@ -24,6 +24,7 @@
"UPDATE wvp_stream_push " +
"SET update_time=#{updateTime}" +
"<if test=\"mediaServerId != null\">, media_server_id=#{mediaServerId}</if>" +
+ "<if test=\"serverId != null\">, server_id=#{serverId}</if>" +
"<if test=\"totalReaderCount != null\">, total_reader_count=#{totalReaderCount}</if>" +
"<if test=\"originType != null\">, origin_type=#{originType}</if>" +
"<if test=\"originTypeStr != null\">, origin_type_str=#{originTypeStr}</if>" +
@@ -89,10 +90,10 @@
@Insert("<script>" +
"Insert INTO wvp_stream_push (app, stream, total_reader_count, origin_type, origin_type_str, " +
- "create_time, alive_second, media_server_id, status, push_ing) " +
+ "create_time, alive_second, media_server_id, server_id, status, push_ing) " +
"VALUES <foreach collection='streamPushItems' item='item' index='index' separator=','>" +
"( #{item.app}, #{item.stream}, #{item.totalReaderCount}, #{item.originType}, " +
- "#{item.originTypeStr},#{item.createTime}, #{item.aliveSecond}, #{item.mediaServerId}, #{item.status} ," +
+ "#{item.originTypeStr},#{item.createTime}, #{item.aliveSecond}, #{item.mediaServerId},#{item.serverId}, #{item.status} ," +
" #{item.pushIng} )" +
" </foreach>" +
"</script>")
diff --git "a/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-mysql-2.6.9.sql" "b/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-mysql-2.6.9.sql"
index 9dab2a4..8eb8d71 100644
--- "a/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-mysql-2.6.9.sql"
+++ "b/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-mysql-2.6.9.sql"
@@ -261,6 +261,7 @@
create_time character varying(50),
alive_second integer,
media_server_id character varying(50),
+ server_id character varying(50),
push_time character varying(50),
status bool default false,
update_time character varying(50),
diff --git "a/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-postgresql-kingbase-2.6.9.sql" "b/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-postgresql-kingbase-2.6.9.sql"
index ed239c3..b48f646 100644
--- "a/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-postgresql-kingbase-2.6.9.sql"
+++ "b/\346\225\260\346\215\256\345\272\223/\345\210\235\345\247\213\345\214\226-postgresql-kingbase-2.6.9.sql"
@@ -261,6 +261,7 @@
create_time character varying(50),
alive_second integer,
media_server_id character varying(50),
+ server_id character varying(50),
push_time character varying(50),
status bool default false,
update_time character varying(50),
diff --git "a/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-mysql-2.6.9.sql" "b/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-mysql-2.6.9.sql"
index 1e4f439..735e76d 100644
--- "a/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-mysql-2.6.9.sql"
+++ "b/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-mysql-2.6.9.sql"
@@ -497,4 +497,7 @@
alter table wvp_media_server
add record_day integer default 7;
+alter table wvp_stream_push
+ add server_id character varying(50);
+
diff --git "a/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-postgresql-kingbase-2.6.9.sql" "b/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-postgresql-kingbase-2.6.9.sql"
index 711f5b6..bbee04e 100644
--- "a/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-postgresql-kingbase-2.6.9.sql"
+++ "b/\346\225\260\346\215\256\345\272\223/\346\233\264\346\226\260-postgresql-kingbase-2.6.9.sql"
@@ -498,5 +498,8 @@
alter table wvp_media_server
add record_day integer default 7;
+alter table wvp_stream_push
+ add server_id character varying(50);
+
--
Gitblit v1.8.0