Merge branch 'main' into main2

648540858

2023-02-10 197f80581b63e0ae116bd06243894e744d522d6c

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

@@ -10,6 +10,8 @@
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;

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

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

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

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

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;

@@ -19,7 +21,10 @@
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.DeferredResultEx;

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

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

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

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

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

@@ -27,6 +32,7 @@
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.util.ObjectUtils;

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

import org.springframework.web.context.request.async.DeferredResult;



import javax.servlet.http.HttpServletRequest;

import javax.sip.InvalidArgumentException;

@@ -35,6 +41,7 @@
import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.UUID;



/**    

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

@@ -78,7 +85,7 @@
   private IStreamProxyService streamProxyService;



   @Autowired

   private IStreamPushService streamPushService;

    private DeferredResultHolder resultHolder;



   @Autowired

   private IMediaService mediaService;

@@ -110,11 +117,10 @@


   /**

    * 服务器定时上报时间，上报间隔可配置，默认10s上报一次

    *

    */

   @ResponseBody

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

   public JSONObject onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param){

    public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {



      logger.info("[ZLM HOOK] 收到zlm心跳：" + param.getMediaServerId());



@@ -129,20 +135,15 @@
      });

      mediaServerService.updateMediaServerKeepalive(param.getMediaServerId(), param.getData());



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

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



      return ret;

        return HookResult.SUCCESS();

   }

   

   /**

    * 播放器鉴权事件，rtsp/rtmp/http-flv/ws-flv/hls的播放都将触发此鉴权事件。

    *  
    */

   @ResponseBody

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

   public JSONObject onPlay(@RequestBody OnPlayHookParam param){

    public HookResult onPlay(@RequestBody OnPlayHookParam param) {

      if (logger.isDebugEnabled()) {

         logger.debug("[ZLM HOOK] 播放鉴权：{}->{}" + param.getMediaServerId(), param);

      }

@@ -158,34 +159,28 @@
            }

         }

      });

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

                return new HookResult(401, "Unauthorized");

         }

      }



      ret.put("code", 0);

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

      return ret;

        return HookResult.SUCCESS();

   }

   

   /**

    * rtsp/rtmp/rtp推流鉴权事件。

    *  
    */

   @ResponseBody

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

   public JSONObject onPublish(@RequestBody OnPublishHookParam param) {

    public HookResultForOnPublish onPublish(@RequestBody OnPublishHookParam param) {



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



      logger.info("[ZLM HOOK]推流鉴权：{}->{}",  param.getMediaServerId(), param);

      JSONObject ret = new JSONObject();



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

      MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);



@@ -194,17 +189,13 @@
            // 推流鉴权

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

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

               ret.put("code", 401);

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

               return ret;

                    return new HookResultForOnPublish(401, "Unauthorized");

            }

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

                    return new HookResultForOnPublish(401, "Unauthorized");

            }

            // 推流自定义播放鉴权码

            String callId = paramMap.get("callId");

@@ -212,9 +203,7 @@
            boolean hasAuthority = userService.checkPushAuthority(callId, sign);

            if (!hasAuthority) {

               logger.info("推流鉴权失败： sign 无权限: callId={}. sign={}", callId, sign);

               ret.put("code", 401);

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

               return ret;

                    return new HookResultForOnPublish(401, "Unauthorized");

            }

            StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);

            streamAuthorityInfo.setCallId(callId);

@@ -232,11 +221,10 @@
         zlmMediaListManager.sendStreamEvent(param.getApp(),param.getStream(), param.getMediaServerId());

      }



      ret.put("code", 0);

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



        HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();

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

         ret.put("enable_audio", true);

            result.setEnable_audio(true);

      }



      taskExecutor.execute(()->{

@@ -245,16 +233,15 @@
            if (mediaInfo != null) {

               subscribe.response(mediaInfo, json);

            }else {

               ret.put("code", 1);

               ret.put("msg", "zlm not register");

                    new HookResultForOnPublish(1, "zlm not register");

            }

         }

      });



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

         ret.put("enable_mp4", userSetting.getRecordSip());

            result.setEnable_mp4(userSetting.getRecordSip());

      }else {

         ret.put("enable_mp4", userSetting.isRecordPushLive());

            result.setEnable_mp4(userSetting.isRecordPushLive());

      }

      List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());

      if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {

@@ -262,25 +249,24 @@
         String channelId = ssrcTransactionForAll.get(0).getChannelId();

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

         if (deviceChannel != null) {

            ret.put("enable_audio", deviceChannel.isHasAudio());

                result.setEnable_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);

                result.setMp4_max_second(10);

                result.setEnable_audio(true);

                result.setEnable_mp4(true);

         }

      }

      return ret;

        return result;

   }

   

   /**

    * rtsp/rtmp流注册或注销时触发此事件；此事件对回复不敏感。

    *  
    */

   @ResponseBody

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

   public JSONObject onStreamChanged(@RequestBody OnStreamChangedHookParam param){

    public HookResult onStreamChanged(@RequestBody OnStreamChangedHookParam param) {



      if (param.isRegist()) {

         logger.info("[ZLM HOOK] 流注册, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

@@ -293,14 +279,13 @@
      ret.put("msg", "success");



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

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

                }

         }

         // 流消失移除redis play

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

@@ -321,7 +306,6 @@
         } else {

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

         }

      });



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

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

@@ -491,22 +475,10 @@
                        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);

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

                                 }

                              }

                           }







                        }

                     } catch (SipException | InvalidArgumentException | ParseException e) {

                                } catch (SipException | InvalidArgumentException | ParseException |

                                         SsrcTransactionNotFoundException e) {

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

                     }

                  }

@@ -514,14 +486,13 @@
            }

         }

      }

        });





      return ret;

        return HookResult.SUCCESS();

   }

   

   /**

    * 流无人观看时事件，用户可以通过此事件选择是否关闭无人看的流。

    *  
    */

   @ResponseBody

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

@@ -632,51 +603,91 @@
   

   /**

    * 流未找到事件，用户可以在此事件触发时，立即去拉流，这样可以实现按需拉流；此事件对回复不敏感。

    *  
    */

   @ResponseBody

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

   public JSONObject onStreamNotFound(@RequestBody OnStreamNotFoundHookParam param){

    public DeferredResult<HookResult> onStreamNotFound(@RequestBody OnStreamNotFoundHookParam param) {

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

      taskExecutor.execute(()->{



        DeferredResult<HookResult> defaultResult = new DeferredResult<>();



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

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

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

            defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));

            return defaultResult;

        }



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

               if (mediaInfo.isRtpEnable()) {

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

                  if (s.length == 2) {

            if (!mediaInfo.isRtpEnable() || s.length != 2) {

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }

                     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 (device == null) {

                defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));

                return defaultResult;

                     }

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

            if (deviceChannel == null) {

                defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));

                return defaultResult;

                  }

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

            RequestMessage msg = new RequestMessage();

            String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;

            boolean exist = resultHolder.exist(key, null);

            msg.setKey(key);

            String uuid = UUID.randomUUID().toString();

            msg.setId(uuid);

            DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());

            DeferredResultEx<HookResult> deferredResultEx = new DeferredResultEx<>(result);



            result.onTimeout(() -> {

                logger.info("点播接口等待超时");

                // 释放rtpserver

                msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));

                resultHolder.invokeResult(msg);

            });

            // TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误

            deferredResultEx.setFilter(result1 -> {

                WVPResult<StreamInfo> wvpResult1 = (WVPResult<StreamInfo>) result1;

                HookResult resultForEnd = new HookResult();

                resultForEnd.setCode(wvpResult1.getCode());

                resultForEnd.setMsg(wvpResult1.getMsg());

                return resultForEnd;

            });



            // 录像查询以channelId作为deviceId查询

            resultHolder.put(key, uuid, deferredResultEx);



            if (!exist) {

                playService.play(mediaInfo, deviceId, channelId, null, eventResult -> {

                    msg.setData(new HookResult(eventResult.statusCode, eventResult.msg));

                    resultHolder.invokeResult(msg);

                }, null);

               }

            return result;

            }else {

               // 拉流代理

               StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());

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

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

               }

            DeferredResult<HookResult> result = new DeferredResult<>();

            result.setResult(HookResult.SUCCESS());

            return result;

            }

         }

      });





      JSONObject ret = new JSONObject();

      ret.put("code", 0);

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

      return ret;

   }

   

   /**

    * 服务器启动事件，可以用于监听服务器崩溃重启；此事件对回复不敏感。

    *  
    */

   @ResponseBody

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

   public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){

    public HookResult onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject) {



      jsonObject.put("ip", request.getRemoteAddr());

      ZLMServerConfig zlmServerConfig = JSON.to(ZLMServerConfig.class, jsonObject);

@@ -692,10 +703,7 @@
         mediaServerService.zlmServerOnline(zlmServerConfig);

      });



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

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

      return ret;

        return HookResult.SUCCESS();

   }



   /**

@@ -703,17 +711,13 @@
    */

   @ResponseBody

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

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

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



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



      JSONObject ret = new JSONObject();

      ret.put("code", 0);

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



      // 查找对应的上级推流，发送停止

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

         return ret;

            return HookResult.SUCCESS();

      }

      taskExecutor.execute(()->{

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

@@ -731,8 +735,7 @@
         }

      });





      return ret;

        return HookResult.SUCCESS();

   }



   /**

@@ -740,12 +743,8 @@
    */

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

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

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



      taskExecutor.execute(()->{

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

@@ -757,7 +756,7 @@
         }

      });



      return ret;

        return HookResult.SUCCESS();

   }



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