使用冒号分隔redis的key

648540858

2024-03-28 1de344674afd6bb35b51b165bbad76dbe6299b7e

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

old mode 100644
new mode 100755
@@ -25,9 +25,12 @@
import com.genersoft.iot.vmp.media.zlm.dto.hook.*;

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

import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;

import com.genersoft.iot.vmp.service.bean.SSRCInfo;

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

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

import com.genersoft.iot.vmp.utils.DateUtil;

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

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

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

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

import org.slf4j.Logger;

@@ -68,9 +71,6 @@


    @Autowired

    private AudioBroadcastManager audioBroadcastManager;



    @Autowired

    private ZLMServerFactory zlmServerFactory;



    @Autowired

    private IPlayService playService;

@@ -115,10 +115,10 @@
    private IUserService userService;



    @Autowired

    private VideoStreamSessionManager sessionManager;

    private ICloudRecordService cloudRecordService;



    @Autowired

    private AssistRESTfulUtils assistRESTfulUtils;

    private VideoStreamSessionManager sessionManager;



    @Autowired

    private SSRCFactory ssrcFactory;

@@ -141,7 +141,7 @@


        taskExecutor.execute(() -> {

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

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

            if (subscribes != null && !subscribes.isEmpty()) {

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

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

    public HookResult onPlay(@RequestBody OnPlayHookParam param) {

        if (logger.isDebugEnabled()) {

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

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

        }

        String mediaServerId = param.getMediaServerId();



@@ -198,8 +198,18 @@


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

        MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);



        if (mediaInfo == null) {

            return new HookResultForOnPublish(200, "success");

        }

        // 推流鉴权的处理

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

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

            if (stream != null) {

                HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();

                result.setEnable_audio(stream.isEnableAudio());

                result.setEnable_mp4(stream.isEnableMp4());

                return result;

            }

            if (userSetting.getPushAuthority()) {

                // 推流鉴权

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

@@ -225,12 +235,6 @@
                streamAuthorityInfo.setSign(sign);

                // 鉴权通过

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

                // 通知assist新的callId

                if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {

                    taskExecutor.execute(() -> {

                        assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);

                    });

                }

            }

        } else {

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

@@ -242,61 +246,80 @@
        taskExecutor.execute(() -> {

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

            if (subscribe != null) {

                if (mediaInfo != null) {

                    subscribe.response(mediaInfo, param);

                } else {

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

                }

                subscribe.response(mediaInfo, param);

            }

        });



        // 是否录像

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

            result.setEnable_mp4(userSetting.getRecordSip());

        } else {

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

        // 国标流

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



                result.setEnable_audio(deviceChannel.isHasAudio());

            }

            // 如果是录像下载就设置视频间隔十秒

            if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {

                result.setMp4_max_second(10);

                result.setEnable_mp4(true);

            }

            // 如果是talk对讲，则默认获取声音

            if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.TALK) {

                result.setEnable_audio(true);

            }

            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());



        }

        if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {

            logger.info("推流时发现尚未设置录像路径，从assist服务中读取");

            JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);

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

                JSONObject dataJson = info.getJSONObject("data");

                if (dataJson != null) {

                    String recordPath = dataJson.getString("record");

                    userSetting.setRecordPath(recordPath);

                    result.setMp4_save_path(recordPath);

                    // 修改zlm中的录像路径

                    if (mediaInfo.isAutoConfig()) {

                        taskExecutor.execute(() -> {

                            mediaServerService.setZLMConfig(mediaInfo, false);

                        });

                    }

            // 单端口模式下修改流 ID

            if (!mediaInfo.isRtpEnable() && inviteInfo == null) {

                String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));

                inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);

                if (inviteInfo != null) {

                    result.setStream_replace(inviteInfo.getStream());

                    logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());

                }

            }



            // 设置音频信息及录制信息

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

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



                // 为录制国标模拟一个鉴权信息, 方便后续写入录像文件时使用

                StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);

                streamAuthorityInfo.setApp(param.getApp());

                streamAuthorityInfo.setStream(ssrcTransactionForAll.get(0).getStream());

                streamAuthorityInfo.setCallId(ssrcTransactionForAll.get(0).getSipTransactionInfo().getCallId());



                redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), ssrcTransactionForAll.get(0).getStream(), streamAuthorityInfo);



                String deviceId = ssrcTransactionForAll.get(0).getDeviceId();

                String channelId = ssrcTransactionForAll.get(0).getChannelId();

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

                if (deviceChannel != null) {

                    result.setEnable_audio(deviceChannel.isHasAudio());

                }

                // 如果是录像下载就设置视频间隔十秒

                if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {

                    // 获取录像的总时长，然后设置为这个视频的时长

                    InviteInfo inviteInfoForDownload = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, param.getStream());

                    if (inviteInfoForDownload != null && inviteInfoForDownload.getStreamInfo() != null) {

                        String startTime = inviteInfoForDownload.getStreamInfo().getStartTime();

                        String endTime = inviteInfoForDownload.getStreamInfo().getEndTime();

                        long difference = DateUtil.getDifference(startTime, endTime) / 1000;

                        result.setMp4_max_second((int) difference);

                        result.setEnable_mp4(true);

                        // 设置为2保证得到的mp4的时长是正常的

                        result.setModify_stamp(2);

                    }

                }

                // 如果是talk对讲，则默认获取声音

                if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.TALK) {

                    result.setEnable_audio(true);

                }

            }

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

            result.setEnable_audio(true);

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

            result.setEnable_audio(true);

        }

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

            String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream();

            OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(receiveKey);

            if (otherRtpSendInfo != null) {

            OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo) redisTemplate.opsForValue().get(receiveKey);



            String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + param.getStream();

            OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo) redisTemplate.opsForValue().get(receiveKeyForPS);

            if (otherRtpSendInfo != null || otherPsSendInfo != null) {

                result.setEnable_mp4(true);

            }

        }

@@ -318,13 +341,10 @@
            logger.info("[ZLM HOOK] 流注销, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

        }



        JSONObject ret = new JSONObject();

        ret.put("code", 0);

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

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

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

        taskExecutor.execute(() -> {

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

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

            if (mediaInfo == null) {

                logger.info("[ZLM HOOK] 流变化未找到ZLM, {}", param.getMediaServerId());

                return;

@@ -335,13 +355,11 @@


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

            // TODO 重构此处逻辑

            boolean isPush = false;

            if (param.isRegist()) {

                // 处理流注册的鉴权信息

                // 处理流注册的鉴权信息， 流注销这里不再删除鉴权信息，下次来了新的鉴权信息会对就的进行覆盖

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

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

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

                    isPush = true;

                    StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());

                    if (streamAuthorityInfo == null) {

                        streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);

@@ -351,10 +369,7 @@
                    }

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

                }

            } else {

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

            }



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

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

                if (param.isRegist()) {

@@ -434,35 +449,40 @@
                } else {

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

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

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

                        if (param.isRegist()) {

                            StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(

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

                            String callId = null;

                            if (streamAuthorityInfo != null) {

                                callId = streamAuthorityInfo.getCallId();

                            }

                            StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaInfo,

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

                            param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));

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

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

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

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

                                param.setSeverId(userSetting.getServerId());

                                zlmMediaListManager.addPush(param);



                        if (mediaServerItem != null) {

                            if (param.isRegist()) {

                                StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());

                                String callId = null;

                                if (streamAuthorityInfo != null) {

                                    callId = streamAuthorityInfo.getCallId();

                                // 冗余数据，自己系统中自用

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

                            }

                        } else {

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

                            OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(

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

                            if (onStreamChangedHookParam != null) {

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

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

                                if ("PUSH".equalsIgnoreCase(type)) {

                                    // 冗余数据，自己系统中自用

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

                                }

                                StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,

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

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

                                    param.setSeverId(userSetting.getServerId());

                                    zlmMediaListManager.addPush(param);

                                }

                            } else {

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

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

                            }

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

                            if (gbStream != null) {

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

                            }

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

@@ -470,7 +490,7 @@
                        GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());

                        if (gbStream != null) {

                            if (userSetting.isUsePushingAsStatus()) {

                                eventPublisher.catalogEventPublishForStream(null, gbStream, param.isRegist()?CatalogEvent.ON:CatalogEvent.OFF);

                                eventPublisher.catalogEventPublishForStream(null, gbStream, param.isRegist() ? CatalogEvent.ON : CatalogEvent.OFF);

                            }

                        }

                        if (type != null) {

@@ -487,31 +507,38 @@
                }

                if (!param.isRegist()) {

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

                    if (sendRtpItems.size() > 0) {

                    if (!sendRtpItems.isEmpty()) {

                        for (SendRtpItem sendRtpItem : sendRtpItems) {

                            if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) {

                                String platformId = sendRtpItem.getPlatformId();

                                ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);

                                Device device = deviceService.getDevice(platformId);



                                    try {

                                        if (platform != null) {

                                            commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                                            redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),

                                                    sendRtpItem.getCallId(), sendRtpItem.getStream());

                                        } else {

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

                                try {

                                    if (platform != null) {

                                        commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                                        redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),

                                                sendRtpItem.getCallId(), sendRtpItem.getStream());

                                    } else {

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

                                        if (sendRtpItem.getPlayType().equals(InviteStreamType.BROADCAST)

                                                || sendRtpItem.getPlayType().equals(InviteStreamType.TALK)) {

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

                                            if (audioBroadcastCatch != null) {

                                                // 来自上级平台的停止对讲

                                                logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());

                                                audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());

                                            }

                                        }

                                    } catch (SipException | InvalidArgumentException | ParseException |

                                             SsrcTransactionNotFoundException e) {

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

                                    }

                                } catch (SipException | InvalidArgumentException | ParseException |

                                         SsrcTransactionNotFoundException e) {

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

                                }

                            }

                        }

                    }

                }



            }

        });

        return HookResult.SUCCESS();

@@ -542,9 +569,9 @@
                }

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

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

                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(

                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(

                            inviteInfo.getChannelId());

                    if (sendRtpItems.size() > 0) {

                    if (!sendRtpItems.isEmpty()) {

                        for (SendRtpItem sendRtpItem : sendRtpItems) {

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

                            try {

@@ -567,16 +594,21 @@
                Device device = deviceService.getDevice(inviteInfo.getDeviceId());

                if (device != null) {

                    try {

                        // 多查询一次防止已经被处理了

                        InviteInfo info = inviteStreamService.getInviteInfo(inviteInfo.getType(),

                                inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());

                        if (info != null) {

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

                                    inviteInfo.getStream(), null);

                        } else {

                            logger.info("[无人观看] 未找到设备的点播信息： {}， 流：{}", inviteInfo.getDeviceId(), param.getStream());

                        }

                    } catch (InvalidArgumentException | ParseException | SipException |

                             SsrcTransactionNotFoundException e) {

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

                    }

                } else {

                    logger.info("[无人观看] 未找到设备： {}，流：{}", inviteInfo.getDeviceId(), param.getStream());

                }



                inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),

@@ -585,7 +617,7 @@
                return ret;

            }

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

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

            if (sendRtpItem != null && "talk".equals(sendRtpItem.getApp())) {

                ret.put("close", false);

                return ret;

            }

@@ -641,14 +673,14 @@


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

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

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

            if ((s.length != 2 && s.length != 4)) {

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }

            String deviceId = s[0];

            String channelId = s[1];

            Device device = redisCatchStorage.getDevice(deviceId);

            if (device == null) {

            if (device == null || !device.isOnLine()) {

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

                return defaultResult;

            }

@@ -657,33 +689,80 @@
                defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));

                return defaultResult;

            }

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

            if (s.length == 2) {

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

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



            result.onTimeout(() -> {

                logger.info("[ZLM HOOK] 自动点播, 等待超时");

                // 释放rtpserver

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

                resultHolder.invokeResult(msg);

            });

                result.onTimeout(() -> {

                    logger.info("[ZLM HOOK] 预览流自动点播, 等待超时");

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

                    resultHolder.invokeAllResult(msg);

                    inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);

                    storager.stopPlay(deviceId, channelId);

                });



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

            resultHolder.put(key, uuid, result);

                resultHolder.put(key, uuid, result);



            if (!exist) {

                playService.play(mediaInfo, deviceId, channelId, null, (code, message, data) -> {

                    msg.setData(new HookResult(code, message));

                if (!exist) {

                    playService.play(mediaInfo, deviceId, channelId, null, (code, message, data) -> {

                        msg.setData(new HookResult(code, message));

                        resultHolder.invokeResult(msg);

                    });

                }

                return result;

            } else if (s.length == 4) {

                // 此时为录像回放， 录像回放格式为> 设备ID_通道ID_开始时间_结束时间

                String startTimeStr = s[2];

                String endTimeStr = s[3];

                if (startTimeStr == null || endTimeStr == null || startTimeStr.length() != 14 || endTimeStr.length() != 14) {

                    defaultResult.setResult(HookResult.SUCCESS());

                    return defaultResult;

                }

                String startTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(startTimeStr);

                String endTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(endTimeStr);

                logger.info("[ZLM HOOK] 回放流未找到, 发起自动点播：{}->{}->{}/{}-{}-{}",

                        param.getMediaServerId(), param.getSchema(),

                        param.getApp(), param.getStream(),

                        startTime, endTime

                );

                RequestMessage msg = new RequestMessage();

                String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + 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());



                result.onTimeout(() -> {

                    logger.info("[ZLM HOOK] 回放流自动点播, 等待超时");

                    // 释放rtpserver

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

                    resultHolder.invokeResult(msg);

                });



                resultHolder.put(key, uuid, result);



                if (!exist) {

                    SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaInfo, param.getStream(), null,

                            device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam());

                    playService.playBack(mediaInfo, ssrcInfo, deviceId, channelId, startTime, endTime, (code, message, data) -> {

                        msg.setData(new HookResult(code, message));

                        resultHolder.invokeResult(msg);

                    });

                }

                return result;

            } else {

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }

            return result;



        } else {

            // 拉流代理

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

@@ -709,7 +788,7 @@
        logger.info("[ZLM HOOK] zlm 启动 " + zlmServerConfig.getGeneralMediaServerId());

        taskExecutor.execute(() -> {

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

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

            if (subscribes != null && !subscribes.isEmpty()) {

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, zlmServerConfig);

                }

@@ -758,14 +837,13 @@
     */

    @ResponseBody

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

    public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam

    public HookResult onRtpServerTimeout(@RequestBody OnRtpServerTimeoutHookParam

            param) {

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



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

            if (subscribes != null && !subscribes.isEmpty()) {

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

@@ -775,6 +853,28 @@
        return HookResult.SUCCESS();

    }



    /**

     * 录像完成事件

     */

    @ResponseBody

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

    public HookResult onRecordMp4(HttpServletRequest request, @RequestBody OnRecordMp4HookParam param) {

        logger.info("[ZLM HOOK] 录像完成事件：{}->{}", param.getMediaServerId(), param.getFile_path());



        taskExecutor.execute(() -> {

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

            if (subscribes != null && !subscribes.isEmpty()) {

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

            }

            cloudRecordService.addRecord(param);



        });



        return HookResult.SUCCESS();

    }



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

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

        if (ObjectUtils.isEmpty(params)) {