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 12d79cb2..74890be7 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 @@ public class WvpRedisMsg { * 消息的ID */ 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 @@ public class WvpRedisMsg { 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 @@ public class WvpRedisMsg { 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 2b0f366b..31966131 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 @@ public class RedisGbPlayMsgListener implements MessageListener { 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 @@ public class RedisGbPlayMsgListener implements MessageListener { 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 @@ public class RedisGbPlayMsgListener implements MessageListener { 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 @@ public class RedisGbPlayMsgListener implements MessageListener { 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 @@ public class RedisGbPlayMsgListener implements MessageListener { WVPResult 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 @@ public class RedisGbPlayMsgListener implements MessageListener { 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 @@ public class RedisGbPlayMsgListener implements MessageListener { 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 class RedisGbPlayMsgListener implements MessageListener { 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 be0924ab..682f07c6 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 @@ import java.util.List; 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 @@ public interface StreamPushMapper { "UPDATE wvp_stream_push " + "SET update_time=#{updateTime}" + ", media_server_id=#{mediaServerId}" + + ", server_id=#{serverId}" + ", total_reader_count=#{totalReaderCount}" + ", origin_type=#{originType}" + ", origin_type_str=#{originTypeStr}" + @@ -89,10 +90,10 @@ public interface StreamPushMapper { @Insert("") diff --git a/数据库/2.6.9/更新-mysql-2.6.9.sql b/数据库/2.6.9/更新-mysql-2.6.9.sql index 1e4f439d..735e76d4 100644 --- a/数据库/2.6.9/更新-mysql-2.6.9.sql +++ b/数据库/2.6.9/更新-mysql-2.6.9.sql @@ -497,4 +497,7 @@ alter table wvp_media_server 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/数据库/2.6.9/更新-postgresql-kingbase-2.6.9.sql b/数据库/2.6.9/更新-postgresql-kingbase-2.6.9.sql index 711f5b6f..bbee04ed 100644 --- a/数据库/2.6.9/更新-postgresql-kingbase-2.6.9.sql +++ b/数据库/2.6.9/更新-postgresql-kingbase-2.6.9.sql @@ -498,5 +498,8 @@ alter table wvp_media_server 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/数据库/初始化-mysql.sql b/数据库/初始化-mysql.sql index 9dab2a45..8eb8d716 100644 --- a/数据库/初始化-mysql.sql +++ b/数据库/初始化-mysql.sql @@ -261,6 +261,7 @@ create table wvp_stream_push ( 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/数据库/初始化-postgresql-kingbase.sql b/数据库/初始化-postgresql-kingbase.sql index ed239c38..b48f6468 100644 --- a/数据库/初始化-postgresql-kingbase.sql +++ b/数据库/初始化-postgresql-kingbase.sql @@ -261,6 +261,7 @@ create table wvp_stream_push ( 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),