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

648540858

2022-05-16 39efda779fcd5d613870e9024cda69da28e571b2

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

@@ -29,6 +29,7 @@
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
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;
@@ -49,7 +50,7 @@
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.util.Date;
import java.util.List;
import java.util.Vector;
@@ -112,6 +113,7 @@

   @Autowired
   private SipConfig config;



   @Override
@@ -200,16 +202,16 @@

            Long startTime = null;
            Long stopTime = null;
            Date start = null;
            Date end = null;
            Instant start = null;
            Instant end = null;
            if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) {
               TimeDescriptionImpl timeDescription = (TimeDescriptionImpl)(sdp.getTimeDescriptions(false).get(0));
               TimeField startTimeFiled = (TimeField)timeDescription.getTime();
               startTime = startTimeFiled.getStartTime();
               stopTime = startTimeFiled.getStopTime();

               start = new Date(startTime*1000);
               end = new Date(stopTime*1000);
               start = Instant.ofEpochMilli(startTime*1000);
               end = Instant.ofEpochMilli(stopTime*1000);
            }
            //  获取支持的格式
            Vector mediaDescriptions = sdp.getMediaDescriptions(true);
@@ -351,13 +353,12 @@
               sendRtpItem.setApp("rtp");
               if ("Playback".equals(sessionName)) {
                  sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
                  SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true);
                  SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true, true);
                  sendRtpItem.setStreamId(ssrcInfo.getStream());
                  // 写入redis， 超时时回复
                  redisCatchStorage.updateSendRTPSever(sendRtpItem);
                  SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                  playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, format.format(start),
                        format.format(end), null, result -> {
                  playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
                        DateUtil.formatter.format(end), null, result -> {
                        if (result.getCode() != 0){
                           logger.warn("录像回放失败");
                           if (result.getEvent() != null) {
@@ -393,7 +394,7 @@
                     if (mediaServerItem.isRtpEnable()) {
                        streamId = String.format("%s_%s", device.getDeviceId(), channelId);
                     }
                     SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, true);
                     SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, true, false);
                     sendRtpItem.setStreamId(ssrcInfo.getStream());
                     // 写入redis， 超时时回复
                     redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -492,7 +493,6 @@
                        gbStream.getApp(), gbStream.getStream(), channelId,
                        mediaTransmissionTCP);


                  if (sendRtpItem == null) {
                     logger.warn("服务器端口资源不足");
                     responseAck(evt, Response.BUSY_HERE);
@@ -562,25 +562,16 @@
      }
   }

   public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId) throws InvalidArgumentException, ParseException, SipException, SdpException {
   public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId1) throws InvalidArgumentException, ParseException, SipException, SdpException {

      // 兼容奇葩的海康这里使用的不是通道编号而是本平台编号
//      if (channelId.equals(config.getId())) {
//         List<AudioBroadcastCatch> all = audioBroadcastManager.getAll();
//         for (AudioBroadcastCatch audioBroadcastCatch : all) {
//            if (audioBroadcastCatch.getDeviceId().equals(requesterId)) {
//               channelId = audioBroadcastCatch.getChannelId();
//            }
//         }
//      }
//      // 兼容失败
//      if (channelId.equals(config.getId())) {
//         responseAck(evt, Response.BAD_REQUEST);
//         return;
//      }
      // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
      Device device = redisCatchStorage.getDevice(requesterId);

      AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(requesterId, channelId1);
      if (audioBroadcastCatch == null) {
         logger.warn("来自设备的Invite请求非语音广播，已忽略");
         responseAck(evt, Response.FORBIDDEN);
         return;
      }
      Request request = evt.getRequest();
      if (device != null) {
         logger.info("收到设备" + requesterId + "的语音广播Invite请求");
@@ -606,7 +597,6 @@

         // 查看是否支持PS 负载96
         int port = -1;
         //boolean recvonly = false;
         boolean mediaTransmissionTCP = false;
         Boolean tcpActive = null;
         for (int i = 0; i < mediaDescriptions.size(); i++) {
@@ -638,7 +628,6 @@
            responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
            return;
         }
         String sessionName = sdp.getSessionName().getValue();
         String addressStr = sdp.getOrigin().getAddress();
         logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", requesterId, addressStr, port, ssrc);

@@ -649,20 +638,19 @@
            return;
         }
         SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
               device.getDeviceId(), channelId,
               device.getDeviceId(), audioBroadcastCatch.getChannelId(),
               mediaTransmissionTCP);
         sendRtpItem.setTcp(mediaTransmissionTCP);
         if (tcpActive != null) {
            sendRtpItem.setTcpActive(tcpActive);
         }
         if (sendRtpItem == null) {
            logger.warn("服务器端口资源不足");
            responseAck(evt, Response.BUSY_HERE);
            return;
         }

         sendRtpItem.setTcp(mediaTransmissionTCP);
         if (tcpActive != null) {
            sendRtpItem.setTcpActive(tcpActive);
         }
         String app = "broadcast";
         String stream = device.getDeviceId() + "_" + channelId;
         String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId();

         CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
         sendRtpItem.setPlayType(InviteStreamType.PLAY);
@@ -671,6 +659,9 @@
         sendRtpItem.setStatus(1);
         sendRtpItem.setApp(app);
         sendRtpItem.setStreamId(stream);
         sendRtpItem.setPt(8);
         sendRtpItem.setUsePs(false);
         sendRtpItem.setOnlyAudio(true);
         redisCatchStorage.updateSendRTPSever(sendRtpItem);

         // hook监听等待设备推流上来
@@ -682,10 +673,9 @@
         subscribeKey.put("schema", "rtmp");
         subscribeKey.put("mediaServerId", mediaServerItem.getId());
         String finalSsrc = ssrc;
         String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + channelId;
         // 流已经存在时直接推流
         if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) {
            logger.info("发现已经在推流");
            dynamicTask.stop(waiteStreamTimeoutTaskKey);
            sendRtpItem.setStatus(2);
            redisCatchStorage.updateSendRTPSever(sendRtpItem);
            StringBuffer content = new StringBuffer(200);
@@ -706,6 +696,10 @@
            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) {
@@ -714,21 +708,19 @@
               throw new RuntimeException(e);
            }
         }else {
            // 流不存在时监听流上线
            // 设置等待推流的超时; 默认20s
            String finalChannelId = channelId;
            String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId();
            dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{
               logger.info("等待推流超时: {}/{}", app, stream);
               if (audioBroadcastManager.exit(device.getDeviceId(), finalChannelId)) {
                  audioBroadcastManager.del(device.getDeviceId(), finalChannelId);
               }else {
                  // 兼容海康使用了错误的通道ID的情况
                  audioBroadcastManager.delByDeviceId(device.getDeviceId());
               }

               subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
               playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
               // 发送bye
               try {
                  cmder.streamByeCmd((SIPDialog)evt.getServerTransaction().getDialog(), (SIPRequest) evt.getRequest(), null);
                  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);
@@ -737,37 +729,43 @@

            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
                  (MediaServerItem mediaServerItemInUse, JSONObject json)->{
                     sendRtpItem.setStatus(2);
                     redisCatchStorage.updateSendRTPSever(sendRtpItem);
                     StringBuffer content = new StringBuffer(200);
                     content.append("v=0\r\n");
                     content.append("o="+ finalChannelId +" 0 0 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");
                     content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                     content.append("a=sendonly\r\n");
                     content.append("a=rtpmap:8 PCMA/8000\r\n");
                     content.append("y="+ finalSsrc + "\r\n");
                     content.append("f=v/////a/1/8/1\r\n");
               logger.info("收到语音对讲推流");
               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");
                  content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
                  content.append("a=sendonly\r\n");
                  content.append("a=rtpmap:8 PCMA/8000\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);
                     } catch (SipException e) {
                        throw new RuntimeException(e);
                     } catch (InvalidArgumentException e) {
                        throw new RuntimeException(e);
                     } catch (ParseException e) {
                        throw new RuntimeException(e);
                     }
                  });
                  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 timeOutTaskKey = "audio-broadcast-" + device.getDeviceId() + channelId;
         dynamicTask.stop(timeOutTaskKey);
         String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
         WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
         wvpResult.setCode(0);