Merge branch 'main' into main2

648540858

2023-02-10 ec95a429859e2819772231e70dcd80cb976c81d2

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

@@ -1,41 +1,40 @@
package com.genersoft.iot.vmp.media.zlm;



import java.text.ParseException;

import java.util.HashMap;

import java.util.List;

import java.util.Map;



import com.alibaba.fastjson2.JSON;

import com.alibaba.fastjson2.JSONObject;

import com.genersoft.iot.vmp.common.StreamInfo;

import com.genersoft.iot.vmp.conf.UserSetting;

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.media.zlm.dto.*;

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

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

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

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

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

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

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

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.StreamContent;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

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

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

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.util.ObjectUtils;

import org.springframework.web.bind.annotation.PostMapping;

import org.springframework.web.bind.annotation.RequestBody;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.ResponseBody;

import org.springframework.web.bind.annotation.RestController;



import com.alibaba.fastjson2.JSONObject;

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

import org.springframework.web.bind.annotation.*;



import javax.servlet.http.HttpServletRequest;

import javax.sip.InvalidArgumentException;

import javax.sip.SipException;

import java.text.ParseException;

import java.util.HashMap;

import java.util.List;

import java.util.Map;



/**    

 * @description:针对 ZLMediaServer的hook事件监听

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



   @Autowired

   private SIPCommanderFroPlatform commanderFroPlatform;

   private ISIPCommanderForPlatform commanderFroPlatform;



   @Autowired

   private AudioBroadcastManager audioBroadcastManager;



   @Autowired

   private ZLMRTPServerFactory zlmrtpServerFactory;



   @Autowired

   private IPlayService playService;

@@ -186,7 +191,7 @@


      if (!"rtp".equals(param.getApp())) {

         if (userSetting.getPushAuthority()) {

// 推流鉴权

            // 推流鉴权

            if (param.getParams() == null) {

               logger.info("推流鉴权失败： 缺少不要参数：sign=md5(user表的pushKey)");

               ret.put("code", 401);

@@ -283,15 +288,19 @@
         logger.info("[ZLM HOOK] 流注销, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

      }



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

      ret.put("msg", "success");



      JSONObject json = (JSONObject) JSON.toJSON(param);

      taskExecutor.execute(()->{

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

      if (mediaInfo == null) {

         return ret;

      }

      taskExecutor.execute(()-> {

         ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);

         if (subscribe != null ) {

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

            if (mediaInfo != null) {

               subscribe.response(mediaInfo, json);

            }

            subscribe.response(mediaInfo, json);

         }

         // 流消失移除redis play

         List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();

@@ -303,43 +312,127 @@
               StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());

               if (streamAuthorityInfo == null) {

                  streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);

               }else {

               } else {

                  streamAuthorityInfo.setOriginType(param.getOriginType());

                  streamAuthorityInfo.setOriginTypeStr(param.getOriginTypeStr());

               }

               redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);

            }

         }else {

         } else {

            redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());

         }

      });



         if ("rtsp".equals(param.getSchema())){

            if (param.isRegist()) {

               mediaServerService.addCount(param.getMediaServerId());

            }else {

               mediaServerService.removeCount(param.getMediaServerId());

      if ("rtsp".equals(param.getSchema())){

         logger.info("流变化：注册->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());

         if (param.isRegist()) {

            mediaServerService.addCount(param.getMediaServerId());

         }else {

            mediaServerService.removeCount(param.getMediaServerId());

         }

         if (param.getOriginType() == OriginType.PULL.ordinal()

               || param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {

            // 设置拉流代理上线/离线

            streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());

         }

         if ("rtp".equals(param.getApp()) && !param.isRegist() ) {

            StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());

            if (streamInfo!=null){

               redisCatchStorage.stopPlay(streamInfo);

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

            }else{

               streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);

               if (streamInfo != null) {

                  redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),

                        streamInfo.getStream(), null);

               }

            }

            if (param.getOriginType() == OriginType.PULL.ordinal()

                  || param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {

               // 设置拉流代理上线/离线

               streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());

         }else if ("broadcast".equals(param.getApp())){

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

            if (param.isRegist() && param.getStream().indexOf("_") > 0) {

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

               if (streamArray.length == 2) {

                  String deviceId = streamArray[0];

                  String channelId = streamArray[1];

                  Device device = deviceService.getDevice(deviceId);

                  if (device != null) {

                     if (param.isRegist()) {

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

                           // 直接推流

                           SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);

                           if (sendRtpItem == null) {

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

                           }else {

                              JSONObject jsonObject = zlmrtpServerFactory.startSendRtp(mediaInfo, sendRtpItem);

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

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

                              }else {

                                 logger.info("[语音喊话] 推流失败, 结果： {}", jsonObject);

                              }



                           }

                        }else {

                           // 开启语音喊话通道

                           try {

                              playService.audioBroadcastCmd(device, channelId, mediaInfo, param.getApp(), param.getStream(),60, false, (msg)->{

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

                              });

                           } catch (InvalidArgumentException | ParseException | SipException e) {

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

                           }

                        }

                     }else {

                        // 流注销

                        playService.stopAudioBroadcast(deviceId, channelId);

                     }

                  } else{

                     logger.info("[语音对讲] 未找到设备：{}", deviceId);

                  }

               }else {

                  logger.info("[语音喊话] 推流格式有误, 格式为： broadcast/设备编号_通道编号 ");

               }

            }

            if ("rtp".equals(param.getApp()) && !param.isRegist() ) {

               StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());

               if (streamInfo!=null){

                  redisCatchStorage.stopPlay(streamInfo);

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

               }else{

                  streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);

                  if (streamInfo != null) {

                     redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),

                           streamInfo.getStream(), null);

         }else if ("talk".equals(param.getApp())){

            // 语音喊话推流  stream需要满足格式deviceId_channelId

            if (param.isRegist() && param.getStream().indexOf("_") > 0) {

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

               if (streamArray.length == 2) {

                  String deviceId = streamArray[0];

                  String channelId = streamArray[1];

                  Device device = deviceService.getDevice(deviceId);

                  if (device != null) {

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

                     if (deviceChannel != null) {

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

                           // 直接推流

                           SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);

                           if (sendRtpItem == null) {

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

                           }else {

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

                              zlmrtpServerFactory.startSendRtp(mediaInfo, sendRtpItem);

                           }

                        }else {

                           // 开启语音喊话通道

                           MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());

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

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

                           }, eventResult -> {

                              System.out.println(eventResult.msg);

                           }, ()->{

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

                           });

                        }



                     }

                  }

               }

            }else {

               if (!"rtp".equals(param.getApp())){

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

                  MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());

            }



         }else{

            if (!"rtp".equals(param.getApp())){

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

               MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());



                  if (mediaServerItem != null){

                     if (param.isRegist()) {

@@ -349,8 +442,8 @@
                           callId = streamAuthorityInfo.getCallId();

                        }

                        StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,

                              param.getApp(), param.getStream(), tracks, callId);

                        param.setStreamInfo(streamInfoByAppAndStream);

                              param.getApp(), param.getStream(), param.getTracks(), callId);

                        param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));

                        redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);

                        if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()

                              || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()

@@ -360,14 +453,15 @@
                        }

                     }else {

                        // 兼容流注销时类型从redis记录获取

                        OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(param.getApp(), param.getStream(), param.getMediaServerId());

                        OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(

                              param.getApp(), param.getStream(), param.getMediaServerId());

                        if (onStreamChangedHookParam != null) {

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

                           redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());

                        }

                        GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());

                        if (gbStream != null) {

//                        eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);

//                           eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);

                        }

                        zlmMediaListManager.removeMedia(param.getApp(), param.getStream());

                     }

@@ -393,25 +487,35 @@
                        ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);

                        Device device = deviceService.getDevice(platformId);



                        try {

                           if (platform != null) {

                              commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                           }else {

                              cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());

                     try {

                        if (platform != null) {

                           commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                        }else {

                           if (sendRtpItem.isOnlyAudio()) {

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

                              if (audioBroadcastCatch != null) {

//                                 playService.stopAudioBroadcast(device.getDeviceId(), sendRtpItem.getChannelId());

                                 if ("talk".equals(param.getApp())) {

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

                                 }else {

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

                                 }

                              }

                           }

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

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







                        }

                     } catch (SipException | InvalidArgumentException | ParseException e) {

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

                     }

                  }

               }

            }

         }

      });

      }



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

      ret.put("msg", "success");



      return ret;

   }

   

@@ -426,7 +530,7 @@
      logger.info("[ZLM HOOK]流无人观看：{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

      JSONObject ret = new JSONObject();

      ret.put("code", 0);

      // 录像下载

      // 国标类型的流

      if ("rtp".equals(param.getApp())){

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

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

@@ -439,6 +543,9 @@
               if (sendRtpItems.size() > 0) {

                  for (SendRtpItem sendRtpItem : sendRtpItems) {

                     ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());

                     if (parentPlatform == null) {

                        continue;

                     }

                     try {

                        commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());

                     } catch (SipException | InvalidArgumentException | ParseException e) {

@@ -508,6 +615,7 @@
               // 修改数据

               streamProxyService.stop(param.getApp(), param.getStream());

            }else {

               // 无人观看不做处理

               ret.put("close", false);

            }

            return ret;

@@ -529,7 +637,7 @@
   @ResponseBody

   @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")

   public JSONObject onStreamNotFound(@RequestBody OnStreamNotFoundHookParam param){

      logger.info("[ZLM HOOK] 流未找到：{}->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

      logger.info("[ZLM HOOK] 流未找到：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

      taskExecutor.execute(()->{

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

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

@@ -597,7 +705,7 @@
   @PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8")

   public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody OnSendRtpStoppedHookParam param){



      logger.info("[ZLM HOOK] 发送rtp被动关闭：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());

      logger.info("[ZLM HOOK] rtp发送关闭：{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

@@ -627,6 +735,31 @@
      return ret;

   }



   /**

    * rtpServer收流超时

    */

   @ResponseBody

   @PostMapping(value = "/on_rtp_server_timeout", produces = "application/json;charset=UTF-8")

   public JSONObject onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam param){

      logger.info("[ZLM HOOK] rtpServer rtp超时：{}->{}({})", param.getMediaServerId(), param.getStream_id(), param.getSsrc());



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

      ret.put("msg", "success");



      taskExecutor.execute(()->{

         JSONObject json = (JSONObject) JSON.toJSON(param);

         List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_rtp_server_timeout);

         if (subscribes != null  && subscribes.size() > 0) {

            for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

               subscribe.response(null, json);

            }

         }

      });



      return ret;

   }



   private Map<String, String> urlParamToMap(String params) {

      HashMap<String, String> map = new HashMap<>();

      if (ObjectUtils.isEmpty(params)) {