优化语音广播流程

648540858

2022-05-10 a9bdb0a706e10d2dffb50ae5a8086dd744bbd976

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

@@ -4,8 +4,11 @@
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
@@ -13,6 +16,10 @@
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import org.ehcache.shadow.org.terracotta.offheapstore.storage.IntegerStorageEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -22,15 +29,18 @@
import javax.sip.Dialog;
import javax.sip.DialogState;
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;
import java.text.ParseException;
import java.util.*;

/**
 * SIP命令类型： ACK请求
 * @author lin
 */
@Component
public class AckRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor {
@@ -51,6 +61,9 @@
    private IRedisCatchStorage redisCatchStorage;

   @Autowired
   private IVideoManagerStorage storager;

   @Autowired
   private ZLMRTPServerFactory zlmrtpServerFactory;

   @Autowired
@@ -61,6 +74,15 @@

   @Autowired
   private DynamicTask dynamicTask;

   @Autowired
   private ISIPCommander cmder;

   @Autowired
   private ISIPCommanderForPlatform commanderForPlatform;

   @Autowired
   private AudioBroadcastManager audioBroadcastManager;


   /**   
@@ -78,10 +100,11 @@
      if (dialog.getState()== DialogState.CONFIRMED) {
         String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
         logger.info("ACK请求： platformGbId->{}", platformGbId);
         ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId);
         // 取消设置的超时任务
         dynamicTask.stop(callIdHeader.getCallId());
         String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
         SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId());
//         String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
         SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
         String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
         MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
         logger.info("收到ACK，开始向上级推流 rtp/{}", sendRtpItem.getStreamId());
@@ -96,9 +119,35 @@
         param.put("src_port", sendRtpItem.getLocalPort());
         param.put("pt", sendRtpItem.getPt());
         param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
         param.put("only_audio", sendRtpItem.isOnlyAudio());
         zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);

         param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
         JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
         if (jsonObject == null) {
            logger.error("RTP推流失败: 请检查ZLM服务");
         } else if (jsonObject.getInteger("code") == 0) {
            if (sendRtpItem.isOnlyAudio()) {
               AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
               audioBroadcastCatch.setStatus(AudioBroadcastCatchStatus.Ok);
               audioBroadcastCatch.setDialog((SIPDialog) evt.getDialog());
               audioBroadcastCatch.setRequest((SIPRequest) evt.getRequest());
               audioBroadcastManager.update(audioBroadcastCatch);
            }
            logger.info("RTP推流成功[ {}/{} ]，{}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
         } else {
            logger.error("RTP推流失败: {}, 参数：{}",jsonObject.getString("msg"),JSONObject.toJSON(param));
            if (sendRtpItem.isOnlyAudio()) {
               // 语音对讲
               try {
                  cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest)evt.getRequest(), null);
               } catch (SipException e) {
                  throw new RuntimeException(e);
               } catch (ParseException e) {
                  throw new RuntimeException(e);
               }
            }else {
               // 向上级平台
               commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());
            }
         }


//         if (streamInfo == null) { // 流还没上来，对方就回复ack