Merge pull request #645 from IKangXu/wvp-28181-2.0

648540858

2022-10-18 1af77ab5f7c11a4b3d59c1989b51b9fca29679ce

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

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.media.zlm;



import java.text.ParseException;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

@@ -7,6 +8,7 @@
import com.alibaba.fastjson.JSON;

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

@@ -31,6 +33,8 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;



import javax.servlet.http.HttpServletRequest;

import javax.sip.InvalidArgumentException;

import javax.sip.SipException;



/**    

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

@@ -57,6 +61,9 @@


   @Autowired

   private IRedisCatchStorage redisCatchStorage;



   @Autowired

   private IDeviceService deviceService;



   @Autowired

   private IMediaServerService mediaServerService;

@@ -103,7 +110,7 @@
   @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")

   public JSONObject onServerKeepalive(@RequestBody JSONObject json){



      logger.info("[ ZLM HOOK ] on_server_keepalive API调用，参数：" + json.toString());

      logger.info("[ ZLM HOOK ]on_server_keepalive API调用，参数：" + json.toString());

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

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

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

@@ -417,10 +424,11 @@
      String schema = item.getSchema();

      List<MediaItem.MediaTrack> tracks = item.getTracks();

      boolean regist = item.isRegist();

      if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal()

            || item.getOriginType() == OriginType.RTSP_PUSH.ordinal()

            || item.getOriginType() == OriginType.RTC_PUSH.ordinal()) {

         if (regist) {

      if (regist) {

         if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal()

               || item.getOriginType() == OriginType.RTSP_PUSH.ordinal()

               || item.getOriginType() == OriginType.RTC_PUSH.ordinal()) {



            StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);

            if (streamAuthorityInfo == null) {

               streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(item);

@@ -429,9 +437,9 @@
               streamAuthorityInfo.setOriginTypeStr(item.getOriginTypeStr());

            }

            redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo);

         }else {

            redisCatchStorage.removeStreamAuthorityInfo(app, stream);

         }

      }else {

         redisCatchStorage.removeStreamAuthorityInfo(app, stream);

      }



      if ("rtsp".equals(schema)){

@@ -451,15 +459,12 @@
            if (streamInfo!=null){

               redisCatchStorage.stopPlay(streamInfo);

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

               // 如果正在给上级推送，则发送bye



            }else{

               streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null);

               if (streamInfo != null) {

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

                        streamInfo.getStream(), null);

               }

               // 如果正在给上级推送，则发送bye

            }

         }else {

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

@@ -509,6 +514,28 @@
               }

            }

         }

         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.getDevice(platformId);



                     try {

                        if (platform != null) {

                           commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                        }else {

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

                        }

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

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

                     }

                  }

               }

            }

         }

      }



      JSONObject ret = new JSONObject();

@@ -531,9 +558,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())) {

@@ -541,37 +571,61 @@
               if (sendRtpItems.size() > 0) {

                  for (SendRtpItem sendRtpItem : sendRtpItems) {

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

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

                     try {

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

                     } catch (SipException | InvalidArgumentException | ParseException e) {

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

                     }

                     redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),

                           sendRtpItem.getCallId(), sendRtpItem.getStreamId());

                  }

               }

            }

            cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(),

                  streamInfoForPlayCatch.getStream(), null);

            redisCatchStorage.stopPlay(streamInfoForPlayCatch);

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

         }else{

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

            if (streamInfoForPlayBackCatch != null) {

               cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(),

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

               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);

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

            if (device != null) {

               try {

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

                        streamInfoForPlayCatch.getStream(), null);

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

                  logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());

               }

            }



            redisCatchStorage.stopPlay(streamInfoForPlayCatch);

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

            return ret;

         }

         MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);

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

         // 录像回放

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

         if (streamInfoForPlayBackCatch != null ) {

            if (streamInfoForPlayBackCatch.isPause()) {

               ret.put("close", false);

            }else {

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

               if (device != null) {

                  try {

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

                           streamInfoForPlayBackCatch.getStream(), null);

                  } catch (InvalidArgumentException | ParseException | SipException |

                         SsrcTransactionNotFoundException e) {

                     logger.error("[无人观看]回放， 发送BYE失败 {}", e.getMessage());

                  }

               }

               redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(),

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

            }

            return ret;

         }

         // 录像下载

         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()) {

@@ -583,12 +637,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;

   }

   

   /**

@@ -603,18 +666,26 @@
      }

      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);

            }

         }

      }

@@ -678,7 +749,11 @@
      if (sendRtpItems.size() > 0) {

         for (SendRtpItem sendRtpItem : sendRtpItems) {

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

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

            try {

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

            } catch (SipException | InvalidArgumentException | ParseException e) {

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

            }

            redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),

                  sendRtpItem.getCallId(), sendRtpItem.getStreamId());

         }