From e85cef434595073b777cea8338543b93d3956b93 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 17 四月 2024 23:23:42 +0800
Subject: [PATCH] 优化多wvp国标级联推流时缓存的清理

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java |   35 +++++++++++++++++++++++++++--------
 1 files changed, 27 insertions(+), 8 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
index 59ff50c..51c5b69 100755
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
@@ -29,6 +29,7 @@
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
+import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 import com.genersoft.iot.vmp.utils.DateUtil;
@@ -53,10 +54,7 @@
 import javax.sip.message.Response;
 import java.text.ParseException;
 import java.time.Instant;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Random;
-import java.util.Vector;
+import java.util.*;
 
 /**
  * SIP鍛戒护绫诲瀷锛� INVITE璇锋眰
@@ -125,6 +123,9 @@
 
     @Autowired
     private SendRtpPortManager sendRtpPortManager;
+
+    @Autowired
+    private RedisPushStreamResponseListener redisPushStreamResponseListener;
 
 
     @Override
@@ -585,7 +586,13 @@
                     sendRtpItem.setOnlyAudio(false);
                     sendRtpItem.setStatus(0);
                     sendRtpItem.setSessionName(sessionName);
-
+                    // 娓呯悊鍙兘瀛樺湪鐨勭紦瀛橀伩鍏嶇敤鍒版棫鐨勬暟鎹�
+                    List<SendRtpItem> sendRtpItemList = redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, gbStream.getStream());
+                    if (!sendRtpItemList.isEmpty()) {
+                        for (SendRtpItem rtpItem : sendRtpItemList) {
+                            redisCatchStorage.deleteSendRTPServer(rtpItem);
+                        }
+                    }
                     if ("push".equals(gbStream.getStreamType())) {
                         sendRtpItem.setPlayType(InviteStreamType.PUSH);
                         if (streamPushItem != null) {
@@ -759,6 +766,7 @@
         redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
         // 璁剧疆瓒呮椂
         dynamicTask.startDelay(sendRtpItem.getCallId(), () -> {
+            redisRpcService.stopWaitePushStreamOnline(sendRtpItem);
             logger.info("[ app={}, stream={} ] 绛夊緟璁惧寮�濮嬫帹娴佽秴鏃�", sendRtpItem.getApp(), sendRtpItem.getStream());
             try {
                 responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂
@@ -770,7 +778,7 @@
         redisRpcService.waitePushStreamOnline(sendRtpItem, (sendRtpItemFromRedis) -> {
             dynamicTask.stop(sendRtpItem.getCallId());
             if (sendRtpItemFromRedis.getServerId().equals(userSetting.getServerId())) {
-
+                logger.info("[绾ц仈鐐规挱] 绛夊緟鐨勬帹娴佸湪鏈钩鍙颁笂绾� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
                 int localPort = sendRtpPortManager.getNextPort(mediaServerItem);
                 if (localPort == 0) {
                     logger.warn("涓婄骇鐐规椂鍒涘缓sendRTPItem澶辫触锛屽彲鑳芥槸鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�");
@@ -801,7 +809,18 @@
                 // 鍏朵粬骞冲彴鍐呭
                 otherWvpPushStream(sendRtpItemFromRedis, request, platform);
             }
-
+        });
+        // 娣诲姞鍥炲鐨勬嫆缁濇垨鑰呴敊璇殑閫氱煡
+        redisPushStreamResponseListener.addEvent(sendRtpItem.getApp(), sendRtpItem.getStream(), response -> {
+            if (response.getCode() != 0) {
+                dynamicTask.stop(sendRtpItem.getCallId());
+                redisRpcService.stopWaitePushStreamOnline(sendRtpItem);
+                try {
+                    responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg());
+                } catch (SipException | InvalidArgumentException | ParseException e) {
+                    logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鐐规挱鍥炲: {}", e.getMessage());
+                }
+            }
         });
     }
 
@@ -811,7 +830,7 @@
      * 鏉ヨ嚜鍏朵粬wvp鐨勬帹娴�
      */
     private void otherWvpPushStream(SendRtpItem sendRtpItem, SIPRequest request, ParentPlatform platform) {
-        logger.info("[绾ц仈鐐规挱]鐩存挱娴佹潵鑷叾浠栧钩鍙帮紝鍙戦�乺edis娑堟伅");
+        logger.info("[绾ц仈鐐规挱] 鏉ヨ嚜鍏朵粬wvp鐨勬帹娴� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
         sendRtpItem = redisRpcService.getSendRtpItem(sendRtpItem);
         // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
         sendRtpItem.setStatus(1);

--
Gitblit v1.8.0