Merge remote-tracking branch 'origin/wvp-28181-2.0' into commercial

648540858

2022-06-27 26739237e2d93460eb869067a6004bfa63a1bdb8

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

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

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
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.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.IStreamPushService;
@@ -28,8 +38,12 @@
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.SerializeUtils;
import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -44,6 +58,8 @@
import javax.sip.message.Response;
import java.text.ParseException;
import java.time.Instant;
import java.util.Date;
import java.util.List;
import java.util.Vector;

/**
@@ -80,12 +96,21 @@

    @Autowired
    private ISIPCommander commander;
	
   @Autowired
   private AudioBroadcastManager audioBroadcastManager;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Autowired
    private IMediaServerService mediaServerService;

   @Autowired
   private ZLMRESTfulUtils zlmresTfulUtils;

   @Autowired
   private IMediaServerService mediaServerService;

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;
@@ -98,6 +123,16 @@

    @Autowired
    private ZLMMediaListManager mediaListManager;

   @Autowired
   private DeferredResultHolder resultHolder;

   @Autowired
   private ZLMHttpHookSubscribe subscribe;

   @Autowired
   private SipConfig config;



    @Autowired
@@ -758,4 +793,224 @@
            responseAck(evt, Response.BAD_REQUEST);
        }
    }
         CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
         sendRtpItem.setPlayType(InviteStreamType.PLAY);
         sendRtpItem.setCallId(callIdHeader.getCallId());
         sendRtpItem.setPlatformId(requesterId);
         sendRtpItem.setStatus(1);
         sendRtpItem.setApp(app);
         sendRtpItem.setStreamId(stream);
         sendRtpItem.setPt(8);
         sendRtpItem.setUsePs(false);
         sendRtpItem.setOnlyAudio(true);
         redisCatchStorage.updateSendRTPSever(sendRtpItem);

         // hook监听等待设备推流上来
         // 添加订阅
         JSONObject subscribeKey = new JSONObject();
         subscribeKey.put("app", app);
         subscribeKey.put("stream", stream);
         subscribeKey.put("regist", true);
         subscribeKey.put("schema", "rtmp");
         subscribeKey.put("mediaServerId", mediaServerItem.getId());
         String finalSsrc = ssrc;
         // 流已经存在时直接推流
         JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream);
         JSONArray tracks = mediaInfo.getJSONArray("tracks");
         Integer codecId = null;
         if (tracks != null && tracks.size() > 0) {
            for (int i = 0; i < tracks.size(); i++) {
               MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class);
               if (track.getCodecType() == 1) {
                  codecId = track.getCodecId();
                  break;
               }
            }
         }
         if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) {
            logger.info("发现已经在推流");
            sendRtpItem.setStatus(2);
            redisCatchStorage.updateSendRTPSever(sendRtpItem);
            StringBuffer content = new StringBuffer(200);
            content.append("v=0\r\n");
            content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion()  + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
            content.append("s=Play\r\n");
            content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
            content.append("t=0 0\r\n");
            if (codecId == null) {
               if (mediaTransmissionTCP) {
                  content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
               }else {
                  content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
               }

               content.append("a=rtpmap:8 PCMA/8000\r\n");
            }else {
               if (codecId == 4) {
                  if (mediaTransmissionTCP) {
                     content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
                  }else {
                     content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
                  }
                  content.append("a=rtpmap:0 PCMU/8000\r\n");
               }else {
                  if (mediaTransmissionTCP) {
                     content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                  }else {
                     content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                  }
                  content.append("a=rtpmap:8 PCMA/8000\r\n");
               }
            }
            if (sendRtpItem.isTcp()) {
               content.append("a=connection:new\r\n");
               if (!sendRtpItem.isTcpActive()) {
                  content.append("a=setup:active\r\n");
               }else {
                  content.append("a=setup:passive\r\n");
               }
            }
            content.append("a=sendonly\r\n");
            content.append("y="+ finalSsrc + "\r\n");
            content.append("f=v/////a/1/8/1\r\n");

            ParentPlatform parentPlatform = new ParentPlatform();
            parentPlatform.setServerIP(device.getIp());
            parentPlatform.setServerPort(device.getPort());
            parentPlatform.setServerGBId(device.getDeviceId());
            try {
               responseSdpAck(evt, content.toString(), parentPlatform);
               Dialog dialog = evt.getDialog();
               audioBroadcastCatch.setDialog((SIPDialog) dialog);
               audioBroadcastCatch.setRequest((SIPRequest) request);
               audioBroadcastManager.update(audioBroadcastCatch);
            } catch (SipException e) {
               throw new RuntimeException(e);
            } catch (InvalidArgumentException e) {
               throw new RuntimeException(e);
            } catch (ParseException e) {
               throw new RuntimeException(e);
            }
         }else {
            // 流不存在时监听流上线
            // 设置等待推流的超时; 默认20s
            String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId();
            dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{
               logger.info("等待推流超时: {}/{}", app, stream);
               subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
               playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
               // 发送bye
               try {
                  responseAck(evt, Response.BUSY_HERE);
               } catch (SipException e) {
                  throw new RuntimeException(e);
               } catch (InvalidArgumentException e) {
                  throw new RuntimeException(e);
               } catch (ParseException e) {
                  throw new RuntimeException(e);
               }
            }, 20*1000);

            boolean finalMediaTransmissionTCP = mediaTransmissionTCP;
            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
                  (MediaServerItem mediaServerItemInUse, JSONObject json)->{
               logger.info("收到语音对讲推流");
               MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class);
               Integer audioCodecId = null;
               if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) {
                  for (int i = 0; i < mediaItem.getTracks().size(); i++) {
                     MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i);
                     if (mediaTrack.getCodecType() == 1) {
                        audioCodecId = mediaTrack.getCodecId();
                        break;
                     }
                  }
               }

               try {
                  sendRtpItem.setStatus(2);
                  redisCatchStorage.updateSendRTPSever(sendRtpItem);
                  StringBuffer content = new StringBuffer(200);
                  content.append("v=0\r\n");
                  content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion()  + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                  content.append("s=Play\r\n");
                  content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
                  content.append("t=0 0\r\n");
                  if (audioCodecId == null) {
                     if (finalMediaTransmissionTCP) {
                        content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                     }else {
                        content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                     }

                     content.append("a=rtpmap:8 PCMA/8000\r\n");
                  }else {
                     if (audioCodecId == 4) {
                        if (finalMediaTransmissionTCP) {
                           content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
                        }else {
                           content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
                        }
                        content.append("a=rtpmap:0 PCMU/8000\r\n");
                     }else {
                        if (finalMediaTransmissionTCP) {
                           content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
                        }else {
                           content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                        }
                        content.append("a=rtpmap:8 PCMA/8000\r\n");
                     }
                  }
                  content.append("a=sendonly\r\n");
                  if (sendRtpItem.isTcp()) {
                     content.append("a=connection:new\r\n");
                     if (!sendRtpItem.isTcpActive()) {
                        content.append("a=setup:active\r\n");
                     }else {
                        content.append("a=setup:passive\r\n");
                     }
                  }
                  content.append("y="+ finalSsrc + "\r\n");
                  content.append("f=v/////a/1/8/1\r\n");

                  ParentPlatform parentPlatform = new ParentPlatform();
                  parentPlatform.setServerIP(device.getIp());
                  parentPlatform.setServerPort(device.getPort());
                  parentPlatform.setServerGBId(device.getDeviceId());

                  responseSdpAck(evt, content.toString(), parentPlatform);
                  Dialog dialog = evt.getDialog();
                  audioBroadcastCatch.setDialog((SIPDialog) dialog);
                  audioBroadcastCatch.setRequest((SIPRequest) request);
                  audioBroadcastManager.update(audioBroadcastCatch);
               } catch (SipException e) {
                  throw new RuntimeException(e);
               } catch (InvalidArgumentException e) {
                  throw new RuntimeException(e);
               } catch (ParseException e) {
                  throw new RuntimeException(e);
               } catch (SdpParseException e) {
                  throw new RuntimeException(e);
               }
            });
         }
         String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
         WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
         wvpResult.setCode(0);
         wvpResult.setMsg("success");
         AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
         audioBroadcastResult.setApp(app);
         audioBroadcastResult.setStream(stream);
         audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem));
         audioBroadcastResult.setCodec("G.711");
         wvpResult.setData(audioBroadcastResult);
         RequestMessage requestMessage = new RequestMessage();
         requestMessage.setKey(key);
         requestMessage.setData(wvpResult);
         resultHolder.invokeAllResult(requestMessage);
      } else {
         logger.warn("来自无效设备/平台的请求");
         responseAck(evt, Response.BAD_REQUEST);
      }
   }
}