合并主线

648540858

2022-10-17 f6320c7dd3dfd25e98bb63f41d79a33af5009a93

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java

@@ -11,12 +11,16 @@
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;

import com.genersoft.iot.vmp.gb28181.bean.*;

import com.genersoft.iot.vmp.gb28181.event.EventPublisher;

import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;

import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;

import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;

import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;

import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;

import com.genersoft.iot.vmp.media.zlm.dto.*;

import com.genersoft.iot.vmp.service.*;

import com.genersoft.iot.vmp.storager.IRedisCatchStorage;

import com.genersoft.iot.vmp.storager.IVideoManagerStorage;

import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;

import com.genersoft.iot.vmp.vmanager.bean.WVPResult;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

@@ -51,7 +55,13 @@
   private SIPCommander cmder;



   @Autowired

   private SIPCommanderFroPlatform commanderFroPlatform;

   private ISIPCommanderForPlatform commanderFroPlatform;



   @Autowired

   private AudioBroadcastManager audioBroadcastManager;



   @Autowired

   private ZLMRTPServerFactory zlmrtpServerFactory;



   @Autowired

   private IPlayService playService;

@@ -466,7 +476,126 @@
                        streamInfo.getStream(), null);

               }

            }

         }else {

         }else if ("broadcast".equals(app)){

            // 语音对讲推流  stream需要满足格式deviceId_channelId

            if (regist && stream.indexOf("_") > 0) {

               String[] streamArray = stream.split("_");

               if (streamArray.length == 2) {

                  String deviceId = streamArray[0];

                  String channelId = streamArray[1];

                  Device device = deviceService.queryDevice(deviceId);

                  if (device != null) {

                     DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);

                     if (deviceChannel != null) {

                        if (audioBroadcastManager.exit(deviceId, channelId)) {

                           // 直接推流

                           SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, stream, null);

                           if (sendRtpItem == null) {

                              // TODO 可能数据错误，重新开启语音通道

                           }else {

                              String is_Udp = sendRtpItem.isTcp() ? "0" : "1";

                              MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());

                              logger.info("rtp/{}开始向上级推流, 目标={}:{}，SSRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());

                              Map<String, Object> param = new HashMap<>(12);

                              param.put("vhost","__defaultVhost__");

                              param.put("app",sendRtpItem.getApp());

                              param.put("stream",sendRtpItem.getStreamId());

                              param.put("ssrc", sendRtpItem.getSsrc());

                              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() ? "1" : "0");



                              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 && jsonObject.getInteger("code") == 0) {

                                 logger.info("[语音对讲] 自动推流成功, device: {}, channel: {}", deviceId, channelId);

                              }



                           }

                        }else {

                           // 开启语音对讲通道

                           try {

                              playService.audioBroadcastCmd(device, channelId, 60, (msg)->{

                                 logger.info("[语音对讲] 通道建立成功, device: {}, channel: {}", deviceId, channelId);

                              });

                           } catch (InvalidArgumentException | ParseException | SipException e) {

                              logger.error("[命令发送失败] 语音对讲: {}", e.getMessage());

                           }

                        }



                     }

                  }

               }

            }



         }else if ("talk".equals(app)){

            // 语音对讲推流  stream需要满足格式deviceId_channelId

            if (regist && stream.indexOf("_") > 0) {

               String[] streamArray = stream.split("_");

               if (streamArray.length == 2) {

                  String deviceId = streamArray[0];

                  String channelId = streamArray[1];

                  Device device = deviceService.queryDevice(deviceId);

                  if (device != null) {

                     DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);

                     if (deviceChannel != null) {

                        if (audioBroadcastManager.exit(deviceId, channelId)) {

                           // 直接推流

                           SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, stream, null);

                           if (sendRtpItem == null) {

                              // TODO 可能数据错误，重新开启语音通道

                           }else {

                              MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());

                              logger.info("rtp/{}开始向上级推流, 目标={}:{}，SSRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());

                              Map<String, Object> param = new HashMap<>(12);

                              param.put("vhost","__defaultVhost__");

                              param.put("app",sendRtpItem.getApp());

                              param.put("stream",sendRtpItem.getStreamId());

                              param.put("ssrc", sendRtpItem.getSsrc());

                              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() ? "1" : "0");



                              JSONObject jsonObject;

                              if (sendRtpItem.isTcpActive()) {

                                 jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);

                              } else {

                                 param.put("is_udp", sendRtpItem.isTcp() ? "0" : "1");

                                 param.put("dst_url", sendRtpItem.getIp());

                                 param.put("dst_port", sendRtpItem.getPort());

                                 jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);

                              }

                              if (jsonObject != null && jsonObject.getInteger("code") == 0) {

                                 logger.info("[语音对讲] 自动推流成功, device: {}, channel: {}", deviceId, channelId);

                              }

                           }

                        }else {

                           // 开启语音对讲通道

                           MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);

                           playService.talk(mediaServerItem, device, channelId, (mediaServer, jsonObject)->{

                              System.out.println("开始推流");

                           }, eventResult -> {

                              System.out.println(eventResult.msg);

                           }, ()->{

                              System.out.println("超时");

                           });

                        }



                     }

                  }

               }

            }



         }else{

            if (!"rtp".equals(app)){

               String type = OriginType.values()[item.getOriginType()].getType();

               MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);

@@ -514,17 +643,30 @@
               }

            }

         }

         if (!regist) {

         if (!regist ) {

            List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream);

            if (sendRtpItems.size() > 0) {

               for (SendRtpItem sendRtpItem : sendRtpItems) {

                  if (sendRtpItem.getApp().equals(app)) {

                     String platformId = sendRtpItem.getPlatformId();

                     ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);

                     Device device = deviceService.queryDevice(platformId);



                     try {

                        commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                     } catch (SipException | InvalidArgumentException | ParseException e) {

                        if (platform != null) {

                           commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                        }else {

                           if ("talk".equals(app) && sendRtpItem.isOnlyAudio()) {

                              AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());

                              if (device != null && audioBroadcastCatch != null) {

//                                 cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null);

                              }

                           }else {

                              cmder.streamByeCmd(device, sendRtpItem.getChannelId(), stream, sendRtpItem.getCallId());

                           }



                        }

                     } catch (SipException | InvalidArgumentException | ParseException | SsrcTransactionNotFoundException e) {

                        logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());

                     }

                  }

@@ -553,9 +695,12 @@
      String app = json.getString("app");

      JSONObject ret = new JSONObject();

      ret.put("code", 0);

      // 录像下载

      ret.put("close", userSetting.getStreamOnDemand());

      if ("rtp".equals(app)){

         ret.put("close", true);

         // 国标流， 点播/录像回放/录像下载

         StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(streamId);

         // 点播

         if (streamInfoForPlayCatch != null) {

            // 收到无人观看说明流也没有在往上级推送

            if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {

@@ -585,10 +730,15 @@


            redisCatchStorage.stopPlay(streamInfoForPlayCatch);

            storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());

         }else{

            StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null);

            if (streamInfoForPlayBackCatch != null) {

               Device device = deviceService.queryDevice(streamInfoForPlayCatch.getDeviceID());

            return ret;

         }

         // 录像回放

         StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null);

         if (streamInfoForPlayBackCatch != null ) {

            if (streamInfoForPlayBackCatch.isPause()) {

               ret.put("close", false);

            }else {

               Device device = deviceService.queryDevice(streamInfoForPlayBackCatch.getDeviceID());

               if (device != null) {

                  try {

                     cmder.streamByeCmd(device,streamInfoForPlayBackCatch.getChannelId(),

@@ -600,20 +750,19 @@
               }

               redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(),

                     streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);

            }else {

               StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, streamId, null);

               // 进行录像下载时无人观看不断流

               if (streamInfoForDownload != null) {

                  ret.put("close", false);

               }

            }

            return ret;

         }

         MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);

         if (mediaServerItem != null && mediaServerItem.getStreamNoneReaderDelayMS() == -1) {

         // 录像下载

         StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, streamId, null);

         // 进行录像下载时无人观看不断流

         if (streamInfoForDownload != null) {

            ret.put("close", false);

            return ret;

         }

         return ret;

      }else {

         // 非国标流 推流/拉流代理

         // 拉流代理

         StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId);

         if (streamProxyItem != null ) {

            if (streamProxyItem.isEnable_remove_none_reader()) {

@@ -625,12 +774,21 @@
            }else if (streamProxyItem.isEnable_disable_none_reader()) {

               // 无人观看停用

               ret.put("close", true);

               // 修改数据

               streamProxyService.stop(app, streamId);

            }else {

               ret.put("close", false);

            }

            return ret;

         }

         return ret;

         // 推流具有主动性，暂时不做处理

//         StreamPushItem streamPushItem = streamPushService.getPush(app, streamId);

//         if (streamPushItem != null) {

//            // TODO 发送停止

//

//         }

      }

      return ret;

   }

   

   /**

@@ -645,19 +803,27 @@
      }

      String mediaServerId = json.getString("mediaServerId");

      MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);

      if (userSetting.isAutoApplyPlay() && mediaInfo != null && mediaInfo.isRtpEnable()) {

      if (userSetting.isAutoApplyPlay() && mediaInfo != null) {

         String app = json.getString("app");

         String streamId = json.getString("stream");

         if ("rtp".equals(app)) {

            String[] s = streamId.split("_");

            if (s.length == 2) {

               String deviceId = s[0];

               String channelId = s[1];

               Device device = redisCatchStorage.getDevice(deviceId);

               if (device != null) {

                  playService.play(mediaInfo,deviceId, channelId, null, null, null);

            if (mediaInfo.isRtpEnable()) {

               String[] s = streamId.split("_");

               if (s.length == 2) {

                  String deviceId = s[0];

                  String channelId = s[1];

                  Device device = redisCatchStorage.getDevice(deviceId);

                  if (device != null) {

                     playService.play(mediaInfo,deviceId, channelId, null, null, null);

                  }

               }

            }

         }else {

            // 拉流代理

            StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(app, streamId);

            if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnable_disable_none_reader()) {

               streamProxyService.start(app, streamId);

            }

         }

      }