Merge branch 'wvp-28181-2.0'

648540858

2022-09-24 d7a1b94f905c5f28c9c8f2d48c3f9e28ebcf9cc4

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

@@ -2,6 +2,8 @@

import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
@@ -14,9 +16,11 @@
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg;
import com.genersoft.iot.vmp.service.impl.RedisGbPlayMsgListener;
import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
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 com.genersoft.iot.vmp.utils.SerializeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -24,18 +28,18 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.sip.Dialog;
import javax.sip.DialogState;
import javax.sip.*;
import javax.sip.RequestEvent;
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 {
@@ -77,6 +81,9 @@
   private ISIPCommanderForPlatform commanderForPlatform;

   @Autowired
   private AudioBroadcastManager audioBroadcastManager;

   @Autowired
   private RedisGbPlayMsgListener redisGbPlayMsgListener;


@@ -104,9 +111,6 @@
      param.put("app",sendRtpItem.getApp());
      param.put("stream",sendRtpItem.getStreamId());
      param.put("ssrc", sendRtpItem.getSsrc());
      param.put("dst_url",sendRtpItem.getIp());
      param.put("dst_port", sendRtpItem.getPort());
      param.put("is_udp", is_Udp);
      param.put("src_port", sendRtpItem.getLocalPort());
      param.put("pt", sendRtpItem.getPt());
      param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
@@ -116,17 +120,58 @@
         param.put("udp_rtcp_timeout", "1");
      }

      if (mediaInfo == null) {
         RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
               sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
               sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
               sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
         redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, jsonObject->{
            startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
         });
      }else {
         JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
         startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
      JSONObject jsonObject;
      if (sendRtpItem.isTcpActive()) {
         jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
      } else {
         param.put("is_udp", is_Udp);
         param.put("dst_url", sendRtpItem.getIp());
         param.put("dst_port", sendRtpItem.getPort());
         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);
               String waiteStreamTimeoutTaskKey = "waite-stream-" + audioBroadcastCatch.getDeviceId() + audioBroadcastCatch.getChannelId();
               dynamicTask.stop(waiteStreamTimeoutTaskKey);
            }
            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(), sendRtpItem.getChannelId(), (SIPRequest) evt.getRequest(), null);
               } catch (SipException | ParseException | InvalidArgumentException e) {
                  logger.error("[命令发送失败] 停止语音对讲: {}", e.getMessage());
               }
            } else {
               // 向上级平台
               commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());
            }
            if (mediaInfo == null) {
               RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
                     sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
                     sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
                     sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
               redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> {
                  startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, json, param, callIdHeader);
               });
            } else {
               JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
               startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader);
            }


         }
      }
   }
   private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform,