修复多wvp推流时，收到停止消息数据处理失败问题

648540858
1 parent 9b62fbe4
Showing 2 changed files with 19 additions and 12 deletions
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
@@ -13,6 +13,7 @@ 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;
@@ -77,6 +78,9 @@ public class RedisGbPlayMsgListener implements MessageListener {
     @Autowired
     private IMediaServerService mediaServerService;
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
     @Autowired
     private DynamicTask dynamicTask;
@@ -322,6 +326,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
         responseSendItemMsg.setSendRtpItem(sendRtpItem);
         responseSendItemMsg.setMediaServerItem(mediaServerItem);
         result.setData(responseSendItemMsg);
+        redisCatchStorage.updateSendRTPSever(sendRtpItem);
         WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
                 userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
@@ -73,12 +73,20 @@ public class RedisPushStreamCloseResponseListener implements MessageListener {
         MessageForPushChannel pushChannel = JSON.parseObject(message.getBody(), MessageForPushChannel.class);
         StreamPushItem push = streamPushService.getPush(pushChannel.getApp(), pushChannel.getStream());
         if (push != null) {
-            if (redisCatchStorage.isChannelSendingRTP(push.getGbId())) {
-                List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
-                        push.getGbId());
-                if (sendRtpItems.size() > 0) {
-                    for (SendRtpItem sendRtpItem : sendRtpItems) {
-                        ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+            List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
+                    push.getGbId());
+            if (!sendRtpItems.isEmpty()) {
+                for (SendRtpItem sendRtpItem : sendRtpItems) {
+                    ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+                    if (parentPlatform != null) {
+                        redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+                        try {
+                            commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem);
+                        } catch (SipException | InvalidArgumentException | ParseException e) {
+                            logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
+                        }
+                    }
+                    if (push.isSelf()) {
                         // 停止向上级推流
                         String streamId = sendRtpItem.getStreamId();
                         Map<String, Object> param = new HashMap<>();
@@ -90,12 +98,6 @@ public class RedisPushStreamCloseResponseListener implements MessageListener {
                         MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                         redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getCallId(), sendRtpItem.getStreamId());
                         zlmServerFactory.stopSendRtpStream(mediaInfo, param);
-
-                        try {
-                            commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem);
-                        } catch (SipException | InvalidArgumentException | ParseException e) {
-                            logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
-                        }
                         if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
                             MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
                                     sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),