jiang
2022-07-13 fc89b7b517e191fb6c9b66fa86dd05ebab0f54c5
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -1,19 +1,25 @@
package com.genersoft.iot.vmp.media.zlm;
import java.util.HashMap;
import java.util.List;
import java.util.UUID;
import java.util.Map;
import com.alibaba.fastjson.JSON;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
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.IVideoManagerStorager;
import com.genersoft.iot.vmp.vmanager.service.IPlayService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.util.StringUtils;
@@ -29,7 +35,7 @@
import javax.servlet.http.HttpServletRequest;
/**    
 * @Description:针对 ZLMediaServer的hook事件监听
 * @description:针对 ZLMediaServer的hook事件监听
 * @author: swwheihei
 * @date:   2020年5月8日 上午10:46:48     
 */
@@ -39,7 +45,6 @@
   private final static Logger logger = LoggerFactory.getLogger(ZLMHttpHookListener.class);
   @Autowired
   private SIPCommander cmder;
@@ -47,10 +52,25 @@
   private IPlayService playService;
   @Autowired
   private IVideoManagerStorager storager;
   private IVideoManagerStorage storager;
   @Autowired
   private IRedisCatchStorage redisCatchStorage;
   @Autowired
   private IMediaServerService mediaServerService;
   @Autowired
   private IStreamProxyService streamProxyService;
   @Autowired
   private IStreamPushService streamPushService;
   @Autowired
   private IMediaService mediaService;
   @Autowired
   private EventPublisher eventPublisher;
    @Autowired
    private ZLMMediaListManager zlmMediaListManager;
@@ -58,17 +78,39 @@
   @Autowired
   private ZLMHttpHookSubscribe subscribe;
   @Value("${media.autoApplyPlay}")
   private boolean autoApplyPlay;
   @Autowired
   private UserSetting userSetting;
   @Value("${media.ip}")
   private String mediaIp;
   @Autowired
   private IUserService userService;
   @Value("${media.wanIp}")
   private String mediaWanIp;
   @Autowired
   private VideoStreamSessionManager sessionManager;
   @Value("${media.port}")
   private int mediaPort;
   /**
    * 服务器定时上报时间,上报间隔可配置,默认10s上报一次
    *
    */
   @ResponseBody
   @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){
      if (logger.isDebugEnabled()) {
         logger.debug("[ ZLM HOOK ] on_server_keepalive API调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_keepalive);
      if (subscribes != null  && subscribes.size() > 0) {
         for (ZLMHttpHookSubscribe.Event subscribe : subscribes) {
            subscribe.response(null, json);
         }
      }
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
      return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
   }
   /**
    * 流量统计事件,播放器或推流器断开时并且耗用流量超过特定阈值时会触发此事件,阈值通过配置文件general.flowThreshold配置;此事件对回复不敏感。
@@ -79,8 +121,9 @@
   public ResponseEntity<String> onFlowReport(@RequestBody JSONObject json){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_flow_report API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_flow_report API调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
@@ -96,8 +139,9 @@
   public ResponseEntity<String> onHttpAccess(@RequestBody JSONObject json){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_http_access API 调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_http_access API 调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("err", "");
@@ -112,15 +156,36 @@
    */
   @ResponseBody
   @PostMapping(value = "/on_play", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onPlay(@RequestBody JSONObject json){
   public ResponseEntity<String> onPlay(@RequestBody OnPlayHookParam param){
      JSONObject json = (JSONObject)JSON.toJSON(param);
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_play API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_play API调用,参数:" + JSON.toJSONString(param));
      }
      String mediaServerId = param.getMediaServerId();
      ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_play, json);
      if (subscribe != null ) {
         MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
         if (mediaInfo != null) {
            subscribe.response(mediaInfo, json);
         }
      }
      JSONObject ret = new JSONObject();
      if (!"rtp".equals(param.getApp())) {
         Map<String, String> paramMap = urlParamToMap(param.getParams());
         StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
         if (streamAuthorityInfo == null
               || (streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(paramMap.get("callId")))) {
            ret.put("code", 401);
            ret.put("msg", "Unauthorized");
            return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
         }
      }
      ret.put("code", 0);
      ret.put("msg", "success");
      return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
      return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
   }
   
   /**
@@ -129,24 +194,91 @@
    */
   @ResponseBody
   @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onPublish(@RequestBody JSONObject json){
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_publish API调用,参数:" + json.toString());
   public ResponseEntity<String> onPublish(@RequestBody OnPublishHookParam param) {
      JSONObject json = (JSONObject) JSON.toJSON(param);
      logger.info("[ ZLM HOOK ]on_publish API调用,参数:" + json.toString());
      JSONObject ret = new JSONObject();
      if (!"rtp".equals(param.getApp())) {
         // 推流鉴权
         if (param.getParams() == null) {
            logger.info("推流鉴权失败: 缺少不要参数:sign=md5(user表的pushKey)");
            ret.put("code", 401);
            ret.put("msg", "Unauthorized");
            return new ResponseEntity<>(ret.toString(), HttpStatus.OK);
         }
         Map<String, String> paramMap = urlParamToMap(param.getParams());
         String sign = paramMap.get("sign");
         if (sign == null) {
            logger.info("推流鉴权失败: 缺少不要参数:sign=md5(user表的pushKey)");
            ret.put("code", 401);
            ret.put("msg", "Unauthorized");
            return new ResponseEntity<>(ret.toString(), HttpStatus.OK);
         }
         // 推流自定义播放鉴权码
         String callId = paramMap.get("callId");
         // 鉴权配置
         boolean hasAuthority = userService.checkPushAuthority(callId, sign);
         if (!hasAuthority) {
            logger.info("推流鉴权失败: sign 无权限: callId={}. sign={}", callId, sign);
            ret.put("code", 401);
            ret.put("msg", "Unauthorized");
            return new ResponseEntity<>(ret.toString(), HttpStatus.OK);
         }
         StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
         streamAuthorityInfo.setCallId(callId);
         streamAuthorityInfo.setSign(sign);
         // 鉴权通过
         redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
      }
      ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json);
      if (subscribe != null) subscribe.response(json);
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
      ret.put("enableHls", true);
      ret.put("enableMP4", false);
      ret.put("enableRtxp", true);
      return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
      ret.put("enable_hls", true);
      if (!"rtp".equals(param.getApp())) {
         ret.put("enable_audio", true);
      }
      String mediaServerId = json.getString("mediaServerId");
      ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json);
      if (subscribe != null) {
         MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
         if (mediaInfo != null) {
            subscribe.response(mediaInfo, json);
         }else {
            ret.put("code", 1);
            ret.put("msg", "zlm not register");
         }
      }
      if ("rtp".equals(param.getApp())) {
         ret.put("enable_mp4", userSetting.getRecordSip());
      }else {
         ret.put("enable_mp4", userSetting.isRecordPushLive());
      }
      List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
      if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
         String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
         String channelId = ssrcTransactionForAll.get(0).getChannelId();
         DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
         if (deviceChannel != null) {
            ret.put("enable_audio", deviceChannel.isHasAudio());
         }
         // 如果是录像下载就设置视频间隔十秒
         if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.download) {
            ret.put("mp4_max_second", 10);
            ret.put("enable_mp4", true);
            ret.put("enable_audio", true);
         }
      }
      return new ResponseEntity<String>(ret.toString(), HttpStatus.OK);
   }
   /**
    * 录制mp4完成后通知事件;此事件对回复不敏感。
    *  
@@ -156,8 +288,26 @@
   public ResponseEntity<String> onRecordMp4(@RequestBody JSONObject json){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_record_mp4 API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_record_mp4 API调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
      return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
   }
   /**
    * 录制hls完成后通知事件;此事件对回复不敏感。
    *
    */
   @ResponseBody
   @PostMapping(value = "/on_record_ts", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onRecordTs(@RequestBody JSONObject json){
      if (logger.isDebugEnabled()) {
         logger.debug("[ ZLM HOOK ]on_record_ts API调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
@@ -173,8 +323,9 @@
   public ResponseEntity<String> onRtspRealm(@RequestBody JSONObject json){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_rtsp_realm API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_rtsp_realm API调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("realm", "");
@@ -191,8 +342,9 @@
   public ResponseEntity<String> onRtspAuth(@RequestBody JSONObject json){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_rtsp_auth API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_rtsp_auth API调用,参数:" + json.toString());
      }
      String mediaServerId = json.getString("mediaServerId");
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("encrypted", false);
@@ -209,14 +361,17 @@
   public ResponseEntity<String> onShellLogin(@RequestBody JSONObject json){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_shell_login API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_shell_login API调用,参数:" + json.toString());
      }
      // TODO 如果是带有rtpstream则开启按需拉流
      // String app = json.getString("app");
      // String stream = json.getString("stream");
      String mediaServerId = json.getString("mediaServerId");
      ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_shell_login, json);
      if (subscribe != null ) {
         MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
         if (mediaInfo != null) {
            subscribe.response(mediaInfo, json);
         }
      ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json);
      if (subscribe != null) subscribe.response(json);
      }
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
@@ -230,30 +385,122 @@
    */
   @ResponseBody
   @PostMapping(value = "/on_stream_changed", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onStreamChanged(@RequestBody JSONObject json){
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_stream_changed API调用,参数:" + json.toString());
   public ResponseEntity<String> onStreamChanged(@RequestBody MediaItem item){
      logger.info("[ ZLM HOOK ]on_stream_changed API调用,参数:" + JSONObject.toJSONString(item));
      String mediaServerId = item.getMediaServerId();
      JSONObject json = (JSONObject) JSON.toJSON(item);
      ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, json);
      if (subscribe != null ) {
         MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
         if (mediaInfo != null) {
            subscribe.response(mediaInfo, json);
         }
      }
      // 流消失移除redis play
      String app = json.getString("app");
      String streamId = json.getString("stream");
      String schema = json.getString("schema");
      boolean regist = json.getBoolean("regist");
      StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
      if ("rtp".equals(app) && !regist ) {
         if (streamInfo!=null){
            redisCatchStorage.stopPlay(streamInfo);
            storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
         }else{
            streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
            redisCatchStorage.stopPlayback(streamInfo);
      String app = item.getApp();
      String stream = item.getStream();
      String schema = item.getSchema();
      List<MediaItem.MediaTrack> tracks = item.getTracks();
      boolean regist = item.isRegist();
      if (regist) {
         StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
         if (streamAuthorityInfo == null) {
            streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(item);
         }else {
            streamAuthorityInfo.setOriginType(item.getOriginType());
            streamAuthorityInfo.setOriginTypeStr(item.getOriginTypeStr());
         }
         redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo);
      }else {
         if (!"rtp".equals(app) && "rtsp".equals(schema)){
            zlmMediaListManager.updateMediaList();
         redisCatchStorage.removeStreamAuthorityInfo(app, stream);
      }
      if ("rtmp".equals(schema)){
         logger.info("on_stream_changed:注册->{}, app->{}, stream->{}", regist, app, stream);
         if (regist) {
            mediaServerService.addCount(mediaServerId);
         }else {
            mediaServerService.removeCount(mediaServerId);
         }
         if (item.getOriginType() == OriginType.PULL.ordinal()
               || item.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
            // 设置拉流代理上线/离线
            streamProxyService.updateStatus(regist, app, stream);
         }
         if ("rtp".equals(app) && !regist ) {
            StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(stream);
            if (streamInfo!=null){
               redisCatchStorage.stopPlay(streamInfo);
               storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
            }else{
               streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null);
               if (streamInfo != null) {
                  redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
                        streamInfo.getStream(), null);
               }
            }
         }else {
            if (!"rtp".equals(app)){
               String type = OriginType.values()[item.getOriginType()].getType();
               MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
               if (mediaServerItem != null){
                  if (regist) {
                     StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
                     StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
                           app, stream, tracks, streamAuthorityInfo.getCallId());
                     item.setStreamInfo(streamInfoByAppAndStream);
                     redisCatchStorage.addStream(mediaServerItem, type, app, stream, item);
                     if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
                           || item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
                           || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
                        item.setSeverId(userSetting.getServerId());
                        zlmMediaListManager.addPush(item);
                     }
//                     List<GbStream> gbStreams = new ArrayList<>();
//                     if (streamPushItem == null || streamPushItem.getGbId() == null) {
//                        GbStream gbStream = storager.getGbStream(app, streamId);
//                        gbStreams.add(gbStream);
//                     }else {
//                        if (streamPushItem.getGbId() != null) {
//                           gbStreams.add(streamPushItem);
//                        }
//                     }
//                     if (gbStreams.size() > 0) {
//                        eventPublisher.catalogEventPublishForStream(null, gbStreams, CatalogEvent.ON);
//                     }
                  }else {
                     // 兼容流注销时类型从redis记录获取
                     MediaItem mediaItem = redisCatchStorage.getStreamInfo(app, stream, mediaServerId);
                     if (mediaItem != null) {
                        type = OriginType.values()[mediaItem.getOriginType()].getType();
                        redisCatchStorage.removeStream(mediaServerItem.getId(), type, app, stream);
                     }
                     GbStream gbStream = storager.getGbStream(app, stream);
                     if (gbStream != null) {
//                        eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
                     }
                     zlmMediaListManager.removeMedia(app, stream);
                  }
                  if (type != null) {
                     // 发送流变化redis消息
                     JSONObject jsonObject = new JSONObject();
                     jsonObject.put("serverId", userSetting.getServerId());
                     jsonObject.put("app", app);
                     jsonObject.put("stream", stream);
                     jsonObject.put("register", regist);
                     jsonObject.put("mediaServerId", mediaServerId);
                     redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                  }
               }
            }
         }
      }
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
@@ -267,41 +514,59 @@
   @ResponseBody
   @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onStreamNoneReader(@RequestBody JSONObject json){
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_stream_none_reader API调用,参数:" + json.toString());
      }
      logger.info("[ ZLM HOOK ]on_stream_none_reader API调用,参数:" + json.toString());
      String mediaServerId = json.getString("mediaServerId");
      String streamId = json.getString("stream");
      String app = json.getString("app");
      StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      if ("rtp".equals(app)){
         JSONObject ret = new JSONObject();
         ret.put("code", 0);
         ret.put("close", true);
         if (streamInfo != null) {
            if (redisCatchStorage.isChannelSendingRTP(streamInfo.getChannelId())) {
         StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(streamId);
         SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransaction(null, null, null, streamId);
         if (streamInfoForPlayCatch != null) {
            // 如果在给上级推流,也不停止。
            if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
               ret.put("close", false);
            } else {
               cmder.streamByeCmd(streamId);
               redisCatchStorage.stopPlay(streamInfo);
               storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
               cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(),
                     streamInfoForPlayCatch.getStream(), null);
               redisCatchStorage.stopPlay(streamInfoForPlayCatch);
               storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
            }
         }else{
            cmder.streamByeCmd(streamId);
            streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
            redisCatchStorage.stopPlayback(streamInfo);
            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);
               }
            }
         }
         MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
         if (mediaServerItem != null && mediaServerItem.getStreamNoneReaderDelayMS() == -1) {
            ret.put("close", false);
         }
         return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
      }else {
         JSONObject ret = new JSONObject();
         ret.put("code", 0);
         ret.put("close", false);
         StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId);
         if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) {
            ret.put("close", true);
            streamProxyService.del(app, streamId);
            String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url();
            logger.info("[{}/{}]<-[{}] 拉流代理无人观看已经移除",  app, streamId, url);
         }else {
            ret.put("close", false);
         }
         return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
      }
   }
   
   /**
@@ -311,38 +576,31 @@
   @ResponseBody
   @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onStreamNotFound(@RequestBody JSONObject json){
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_stream_not_found API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_stream_not_found API调用,参数:" + json.toString());
      }
      if (autoApplyPlay) {
      String mediaServerId = json.getString("mediaServerId");
      MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
      if (userSetting.isAutoApplyPlay() && mediaInfo != null && mediaInfo.isRtpEnable()) {
         String app = json.getString("app");
         String streamId = json.getString("stream");
            StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
         if ("rtp".equals(app) && streamId.indexOf("gb_play") > -1 && streamInfo == null) {
         if ("rtp".equals(app)) {
            String[] s = streamId.split("_");
            if (s.length == 4) {
               String deviceId = s[2];
               String channelId = s[3];
               Device device = storager.queryVideoDevice(deviceId);
            if (s.length == 2) {
               String deviceId = s[0];
               String channelId = s[1];
               Device device = redisCatchStorage.getDevice(deviceId);
               if (device != null) {
                  UUID uuid = UUID.randomUUID();
                  cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
                     logger.info("收到订阅消息: " + response.toJSONString());
                     playService.onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
                  }, null);
                  playService.play(mediaInfo,deviceId, channelId, null, null, null);
               }
            }
         }
      }
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
      return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
      return new ResponseEntity<>(ret.toString(),HttpStatus.OK);
   }
   
   /**
@@ -351,33 +609,40 @@
    */
   @ResponseBody
   @PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8")
   public ResponseEntity<String> onServerStarted(HttpServletRequest request, @RequestBody JSONObject json){
   public ResponseEntity<String> onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
      
      if (logger.isDebugEnabled()) {
         logger.debug("ZLM HOOK on_server_started API调用,参数:" + json.toString());
         logger.debug("[ ZLM HOOK ]on_server_started API调用,参数:" + jsonObject.toString());
      }
//      String data = json.getString("data");
//      List<MediaServerConfig> mediaServerConfigs = JSON.parseArray(JSON.toJSONString(json), MediaServerConfig.class);
//      MediaServerConfig mediaServerConfig = mediaServerConfigs.get(0);
      String remoteAddr = request.getRemoteAddr();
      jsonObject.put("ip", remoteAddr);
      List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_started);
      if (subscribes != null && subscribes.size() > 0) {
      if (subscribes != null  && subscribes.size() > 0) {
         for (ZLMHttpHookSubscribe.Event subscribe : subscribes) {
            subscribe.response(json);
            subscribe.response(null, jsonObject);
         }
      }
      MediaServerConfig mediaServerConfig = JSON.toJavaObject(json, MediaServerConfig.class);
      mediaServerConfig.setWanIp(StringUtils.isEmpty(mediaWanIp)? mediaIp: mediaWanIp);
      mediaServerConfig.setLocalIP(mediaIp);
      redisCatchStorage.updateMediaInfo(mediaServerConfig);
      // 重新发起代理
      JSONObject ret = new JSONObject();
      ret.put("code", 0);
      ret.put("msg", "success");
      return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
   }
   private Map<String, String> urlParamToMap(String params) {
      HashMap<String, String> map = new HashMap<>();
      if (StringUtils.isEmpty(params)) {
         return map;
      }
      String[] paramsArray = params.split("&");
      if (paramsArray.length == 0) {
         return map;
      }
      for (String param : paramsArray) {
         String[] paramArray = param.split("=");
         if (paramArray.length == 2){
            map.put(paramArray[0], paramArray[1]);
         }
      }
      return map;
   }
}