From 42a2772d1aa7493bcc4fac3e24ee8eda4eebc23d Mon Sep 17 00:00:00 2001
From: xubinbin <1323875150@qq.com>
Date: 星期二, 12 十二月 2023 17:09:04 +0800
Subject: [PATCH] bugfix:请求头带token, SecurityUtils 获取用户id 一直为0 #1195

---
 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java |  243 ++++++++++++++++++++++++++----------------------
 1 files changed, 133 insertions(+), 110 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 61dea12..2b0f366
--- 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.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;
 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,27 +69,21 @@
 
 
     @Autowired
-    private ZLMMediaListManager zlmMediaListManager;
+    private RedisTemplate<Object, Object> redisTemplate;
 
     @Autowired
-    private ZLMRTPServerFactory zlmrtpServerFactory;
+    private ZLMServerFactory zlmServerFactory;
 
     @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;
             });
         }
-
-
-
-
-
-
     }
 
     /**
@@ -236,7 +230,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);
     }
@@ -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,12 +263,12 @@
                     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;
         }
         // 纭畾娴佹槸鍚﹀湪绾�
-        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 {
@@ -292,13 +286,13 @@
                         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());
 
             // 娣诲姞璁㈤槄
             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);
                     });
@@ -306,8 +300,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));
         }
     }
 
@@ -315,7 +311,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());
@@ -331,7 +327,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 +364,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 +389,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