修复多wvp模式推流时信息存储错误

648540858
1 parent c25a99d6
Showing 7 changed files with 27 additions and 18 deletions
src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
数据库/初始化-mysql-2.6.9.sql
数据库/初始化-postgresql-kingbase-2.6.9.sql
数据库/更新-mysql-2.6.9.sql
数据库/更新-postgresql-kingbase-2.6.9.sql
@@ -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;
     }
 }
@@ -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<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 @@ 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);
@@ -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}" +
             "<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 @@ public interface StreamPushMapper {
     @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>")
@@ -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),
@@ -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),
@@ -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);
+
@@ -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);
+