Merge branch 'wvp-28181-2.0' into main-dev

648540858

2023-06-27 5a7a7a12bde268c104610d3e81a08df06ccab5eb

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java

@@ -1,9 +1,11 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
@@ -13,6 +15,7 @@
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
@@ -29,6 +32,7 @@
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
@@ -57,6 +61,9 @@
   private IRedisCatchStorage redisCatchStorage;

   @Autowired
   private IInviteStreamService inviteStreamService;

   @Autowired
   private IDeviceService deviceService;

   @Autowired
@@ -67,6 +74,9 @@

   @Autowired
   private ZLMRTPServerFactory zlmrtpServerFactory;

   @Autowired
   private SSRCFactory ssrcFactory;

   @Autowired
   private IMediaServerService mediaServerService;
@@ -100,92 +110,88 @@
      } catch (SipException | InvalidArgumentException | ParseException e) {
         logger.error("[回复BYE信息失败]，{}", e.getMessage());
      }

      SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, null, request.getCallIdHeader().getCallId());

      if (sendRtpItem != null){
         logger.info("[收到bye] {}/{}", sendRtpItem.getPlatformId(), sendRtpItem.getChannelId());
         String streamId = sendRtpItem.getStream();
         MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId());
         if (mediaServerItem == null) {
            return;
         }

         Boolean ready = zlmrtpServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), streamId);
         if (!ready) {
            logger.info("[收到bye] 发现流{}/{}已经结束，不需处理", sendRtpItem.getApp(), sendRtpItem.getStream());
            return;
         }
         Map<String, Object> param = new HashMap<>();
         param.put("vhost","__defaultVhost__");
         param.put("app",sendRtpItem.getApp());
         param.put("stream",streamId);
         param.put("ssrc",sendRtpItem.getSsrc());
         logger.info("[收到bye] 停止推流：{}", streamId);
         MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
         redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), request.getCallIdHeader().getCallId(), null);
         zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);

         int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
         if (totalReaderCount <= 0) {
            logger.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId);
            if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
               Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
               if (device == null) {
                  logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
      CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
         String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
         String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
         SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId());
         logger.info("[收到bye] {}/{}", platformGbId, channelId);
         if (sendRtpItem != null){
            String streamId = sendRtpItem.getStream();
            Map<String, Object> param = new HashMap<>();
            param.put("vhost","__defaultVhost__");
            param.put("app",sendRtpItem.getApp());
            param.put("stream",streamId);
            param.put("ssrc",sendRtpItem.getSsrc());
            logger.info("[收到bye] 停止向上级推流：{}", streamId);
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
            ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
            zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
            int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
            if (totalReaderCount <= 0) {
               logger.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId);
               if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
                  Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
                  if (device == null) {
                     logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
                  }
                  try {
                     logger.warn("[停止点播] {}/{}", sendRtpItem.getDeviceId(), channelId);
                     cmder.streamByeCmd(device, channelId, streamId, null);
                  } catch (InvalidArgumentException | ParseException | SipException |
                         SsrcTransactionNotFoundException e) {
                     logger.error("[收到bye] {} 无其它观看者，通知设备停止推流， 发送BYE失败 {}",streamId, e.getMessage());
                  }
               }
               try {
                  logger.warn("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
                  cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null);
               } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
                  logger.error("[收到bye] {} 无其它观看者，通知设备停止推流， 发送BYE失败 {}",streamId, e.getMessage());
               if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
                  MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
                        sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
                        sendRtpItem.getPlatformId(), null, null, sendRtpItem.getMediaServerId());
                  redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
               }
            }

            if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
               MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
                     sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
                     sendRtpItem.getPlatformId(), null, null, sendRtpItem.getMediaServerId());
               redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
         }
         // 可能是设备主动停止
         Device device = storager.queryVideoDeviceByChannelId(platformGbId);
         if (device != null) {
            storager.stopPlay(device.getDeviceId(), channelId);
            SsrcTransaction ssrcTransactionForPlay = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
            if (ssrcTransactionForPlay != null){
               if (ssrcTransactionForPlay.getCallId().equals(callIdHeader.getCallId())){
                  // 释放ssrc
                  MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransactionForPlay.getMediaServerId());
                  if (mediaServerItem != null) {
                     mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlay.getSsrc());
                  }
                  streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlay.getStream());
               }
               InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId);
               inviteStreamService.removeInviteInfo(inviteInfo);
               if (inviteInfo != null) {
                  if (inviteInfo.getStreamInfo() != null) {
                     mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream());
                  }
               }
            }
         }

         playService.stopAudioBroadcast(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
      }

      String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
      String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();

      // 可能是设备主动停止
      Device device = storager.queryVideoDeviceByChannelId(platformGbId);
      if (device != null) {
         storager.stopPlay(device.getDeviceId(), channelId);
         StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
         if (streamInfo != null) {
            redisCatchStorage.stopPlay(streamInfo);
            mediaServerService.closeRTPServer(streamInfo.getMediaServerId(), streamInfo.getStream());
         }
         SsrcTransaction ssrcTransactionForPlay = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
         if (ssrcTransactionForPlay != null){
            if (ssrcTransactionForPlay.getCallId().equals(request.getCallIdHeader().getCallId())){
            SsrcTransaction ssrcTransactionForPlayBack = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, callIdHeader.getCallId(), null);
            if (ssrcTransactionForPlayBack != null) {
               // 释放ssrc
               MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransactionForPlay.getMediaServerId());
               MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransactionForPlayBack.getMediaServerId());
               if (mediaServerItem != null) {
                  mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlay.getSsrc());
                  mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlayBack.getSsrc());
               }
               streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlay.getStream());
               streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlayBack.getStream());
               InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, device.getDeviceId(), channelId);

               if (inviteInfo != null) {
                  inviteStreamService.removeInviteInfo(inviteInfo);
                  if (inviteInfo.getStreamInfo() != null) {
                     mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream());
                  }
               }
            }
         }
         SsrcTransaction ssrcTransactionForPlayBack = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, request.getCallIdHeader().getCallId(), null);
         if (ssrcTransactionForPlayBack != null) {
            // 释放ssrc
            MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransactionForPlayBack.getMediaServerId());
            if (mediaServerItem != null) {
               mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlayBack.getSsrc());
            }
            streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlayBack.getStream());
         }
      }

      SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, request.getCallIdHeader().getCallId(), null);
      if (ssrcTransaction != null) {
         // 释放ssrc
@@ -203,29 +209,34 @@
//                  break;
//               case download:
//                  break;
            case broadcast:
               String deviceId = ssrcTransaction.getDeviceId();
            case BROADCAST:
               String channelId1 = ssrcTransaction.getChannelId();

               Device deviceFromTransaction = storager.queryVideoDevice(deviceId);
               Device deviceFromTransaction = storager.queryVideoDevice(ssrcTransaction.getDeviceId());
               if (deviceFromTransaction == null) {
                  ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(deviceId);
                  ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(ssrcTransaction.getDeviceId());
                  if (parentPlatform != null) {
                     // 来自上级平台的停止对讲
                     logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", ssrcTransaction.getDeviceId(), channelId1);
                     // 释放ssrc
                     MediaServerItem mediaServerItemFromTransaction = mediaServerService.getOne(ssrcTransaction.getMediaServerId());
                     if (mediaServerItemFromTransaction != null) {
                        mediaServerService.releaseSsrc(mediaServerItemFromTransaction.getId(), ssrcTransaction.getSsrc());
                     }
                     streamSession.remove(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
                     if (mediaServerItem != null) {
                        zlmrtpServerFactory.closeRtpServer(mediaServerItem, ssrcTransaction.getStream());
                     }
                     // 查找来源的对讲设备，发送停止
                     Device sourceDevice = storager.queryVideoDeviceByPlatformIdAndChannelId(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
                     if (sourceDevice != null) {
                        playService.stopAudioBroadcast(sourceDevice.getDeviceId(), channelId);
                     }
                  }
               }else {
                  // 来自设备的停止对讲

                  // 如果是来自设备，则听停止推流。 来自上级则停止收流
                  AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId1);
                  AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(ssrcTransaction.getDeviceId(), channelId1);
                  if (audioBroadcastCatch != null) {
                     //
                     SendRtpItem sendRtpItemForBroadcast =  redisCatchStorage.querySendRTPServer(deviceId, channelId1,
                     SendRtpItem sendRtpItemForBroadcast =  redisCatchStorage.querySendRTPServer(ssrcTransaction.getDeviceId(), channelId1,
                           audioBroadcastCatch.getStream(), audioBroadcastCatch.getSipTransactionInfo().getCallId());
                     if (sendRtpItemForBroadcast != null) {
                        MediaServerItem mediaServerItemForBroadcast = mediaServerService.getOne(sendRtpItem.getMediaServerId());
@@ -250,13 +261,14 @@
                           List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(null, channelId1, null, null);
                           if (ssrcTransactions.size() > 0) {
                              for (SsrcTransaction transaction : ssrcTransactions) {
                                 if (transaction.getType().equals(VideoStreamSessionManager.SessionType.broadcast)) {
                                 if (transaction.getType().equals(InviteSessionType.BROADCAST)) {
                                    ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(transaction.getDeviceId());
                                    if (parentPlatform != null) {
                                       try {
                                          commanderForPlatform.streamByeCmd(parentPlatform, channelId1, transaction.getStream(), transaction.getCallId(), eventResult -> {
                                             streamSession.remove(transaction.getDeviceId(), transaction.getChannelId(), transaction.getStream());
                                          });
                                          audioBroadcastManager.del(transaction.getDeviceId(), channelId1);
                                       } catch (InvalidArgumentException | SipException | ParseException |
                                              SsrcTransactionNotFoundException e) {
                                          logger.error("[命令发送失败] 向{}发送bye失败", transaction.getDeviceId());
@@ -273,14 +285,13 @@
                           }

                        }
                        redisCatchStorage.deleteSendRTPServer(deviceId, channelId1,
                        redisCatchStorage.deleteSendRTPServer(ssrcTransaction.getDeviceId(), channelId1,
                              audioBroadcastCatch.getStream(), audioBroadcastCatch.getSipTransactionInfo().getCallId());

                     }

                     audioBroadcastManager.del(deviceId, channelId1);
                  }
               }
               audioBroadcastManager.del(ssrcTransaction.getDeviceId(), channelId1);
               break;
            default:
               break;