hook优化

648540858

2024-03-31 4548695a0b79cc6a3cc940d698cdf1d0b535d570

hook优化

 src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java

@@ -6,8 +6,8 @@
import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEvent;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.event.subscribe.mobilePosition.MobilePositionEvent;
import com.genersoft.iot.vmp.media.event.MediaServerOfflineEvent;
import com.genersoft.iot.vmp.media.event.MediaServerOnlineEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOfflineEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOnlineEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;

 src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java

@@ -1,26 +1,13 @@
package com.genersoft.iot.vmp.gb28181.session;

import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.AudioBroadcastCatch;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java

@@ -14,12 +14,11 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderProvider;

import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;

import com.genersoft.iot.vmp.gb28181.utils.SipUtils;

import com.genersoft.iot.vmp.media.event.hook.Hook;

import com.genersoft.iot.vmp.media.event.hook.HookType;

import com.genersoft.iot.vmp.media.service.IMediaServerService;

import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;

import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;

import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;

import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;

import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamPush;

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

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

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

@@ -279,11 +278,11 @@
        }



        logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getSdpIp(), ssrcInfo.getPort());

        HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());

        subscribe.addSubscribe(hookSubscribe, (MediaServer mediaServerItemInUse, HookParam hookParam) -> {

        Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, "rtp", stream, mediaServerItem.getId());

        subscribe.addSubscribe(rtpHook, (hookData) -> {

            if (event != null) {

                event.response(mediaServerItemInUse, hookParam);

                subscribe.removeSubscribe(hookSubscribe);

                event.response(hookData);

                subscribe.removeSubscribe(rtpHook);

            }

        });

        String sdpIp;

@@ -453,13 +452,13 @@
        //ssrc

        content.append("y=" + ssrcInfo.getSsrc() + "\r\n");



        HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());

        Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcInfo.getStream(), mediaServerItem.getId());

        // 添加订阅

        subscribe.addSubscribe(hookSubscribe, (MediaServer mediaServerItemInUse, HookParam hookParam) -> {

        subscribe.addSubscribe(rtpHook, (hookData) -> {

            if (hookEvent != null) {

                hookEvent.response(mediaServerItemInUse, hookParam);

                hookEvent.response(hookData);

            }

            subscribe.removeSubscribe(hookSubscribe);

            subscribe.removeSubscribe(rtpHook);

        });

        Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), SipUtils.getNewViaTag(), SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()), ssrcInfo.getSsrc());



@@ -554,19 +553,18 @@


        content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc

        logger.debug("此时请求下载信令的ssrc===>{}",ssrcInfo.getSsrc());

        HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());

        Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcInfo.getStream(), mediaServerItem.getId());

        // 添加订阅

        CallIdHeader newCallIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());

        String callId= newCallIdHeader.getCallId();

        subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> {

        subscribe.addSubscribe(rtpHook, (hookData) -> {

            logger.debug("sipc 添加订阅===callId {}",callId);

            hookEvent.response(mediaServerItemInUse, hookParam);

            subscribe.removeSubscribe(hookSubscribe);

            hookSubscribe.getContent().put("regist", false);

            hookSubscribe.getContent().put("schema", "rtsp");

            hookEvent.response(hookData);

            subscribe.removeSubscribe(rtpHook);

            // 添加流注销的订阅，注销了后向设备发送bye

            subscribe.addSubscribe(hookSubscribe,

                    (mediaServerItemForEnd, hookParam1) -> {

            Hook departureHook = Hook.getInstance(HookType.on_media_departure, "rtp", ssrcInfo.getStream(), mediaServerItem.getId());

            subscribe.addSubscribe(departureHook,

                    (departureHookData) -> {

                        logger.info("[录像]下载结束， 发送BYE");

                        try {

                            streamByeCmd(device, channelId, ssrcInfo.getStream(), callId);

@@ -604,20 +602,20 @@
        }



        logger.info("[语音喊话] {} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), sendRtpItem.getPort());

        HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());

        subscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItemInUse, hookParam) -> {

        Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", stream, mediaServerItem.getId());

        subscribe.addSubscribe(hook, (hookData) -> {

            if (event != null) {

                event.response(mediaServerItemInUse, hookParam);

                subscribe.removeSubscribe(hookSubscribeForStreamChange);

                event.response(hookData);

                subscribe.removeSubscribe(hook);

            }

        });



        CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());

        callIdHeader.setCallId(callId);

        HookSubscribeForStreamPush hookSubscribeForStreamPush = HookSubscribeFactory.on_publish("rtp", stream,  null, mediaServerItem.getId());

        subscribe.addSubscribe(hookSubscribeForStreamPush, (mediaServerItemInUse, hookParam) -> {

        Hook publishHook = Hook.getInstance(HookType.on_publish, "rtp", stream, mediaServerItem.getId());

        subscribe.addSubscribe(publishHook, (hookData) -> {

            if (eventForPush != null) {

                eventForPush.response(mediaServerItemInUse, hookParam);

                eventForPush.response(hookData);

            }

        });

        //

@@ -1260,7 +1258,6 @@
     * @param startPriority 报警起始级别（可选）

     * @param endPriority   报警终止级别（可选）

     * @param alarmMethod   报警方式条件（可选）

     * @param alarmType     报警类型

     * @param startTime     报警发生起始时间（可选）

     * @param endTime       报警发生终止时间（可选）

     * @return true = 命令发送成功


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -13,9 +13,9 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
@@ -905,11 +905,11 @@
        }

        logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
        HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());
        subscribe.addSubscribe(hookSubscribe, (MediaServer mediaServerItemInUse, HookParam hookParam) -> {
        Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", stream, mediaServerItem.getId());
        subscribe.addSubscribe(hook, (hookData) -> {
            if (event != null) {
                event.response(mediaServerItemInUse, hookParam);
                subscribe.removeSubscribe(hookSubscribe);
                event.response(hookData);
                subscribe.removeSubscribe(hook);
            }
        });
        String sdpIp = mediaServerItem.getSdpIp();
@@ -949,7 +949,7 @@
        sipSender.transmitRequest(sipLayer.getLocalIp(platform.getDeviceIp()), request, (e -> {
            streamSession.remove(platform.getServerGBId(), channelId, ssrcInfo.getStream());
            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
            subscribe.removeSubscribe(hookSubscribe);
            subscribe.removeSubscribe(hook);
            errorEvent.response(e);
        }), e -> {
            ResponseEvent responseEvent = (ResponseEvent) e.event;

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -18,8 +18,8 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
@@ -115,7 +115,7 @@
    private IMediaServerService mediaServerService;

    @Autowired
    private HookSubscribe zlmHttpHookSubscribe;
    private HookSubscribe hookSubscribe;

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;
@@ -723,17 +723,16 @@
            // TODO 控制启用以使设备上线
            logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
            // 监听流上线
            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(gbStream.getApp(), gbStream.getStream(), true, "rtsp", mediaServerItem.getId());
            zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> {
                OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
                logger.info("[上级点播]拉流代理已经就绪， {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream());
            Hook hook = Hook.getInstance(HookType.on_media_arrival, gbStream.getApp(), gbStream.getStream(), mediaServerItem.getId());
            this.hookSubscribe.addSubscribe(hook, (hookData) -> {
                logger.info("[上级点播]拉流代理已经就绪， {}/{}", hookData.getApp(), hookData.getStream());
                dynamicTask.stop(callIdHeader.getCallId());
                pushProxyStream(evt, request, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive,
                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
            });
            dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
                logger.info("[ app={}, stream={} ] 等待拉流代理流超时", gbStream.getApp(), gbStream.getStream());
                zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
                this.hookSubscribe.removeSubscribe(hook);
            }, userSetting.getPlatformPlayTimeout());
            boolean start = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
            if (!start) {
@@ -742,7 +741,7 @@
                } catch (SipException | InvalidArgumentException | ParseException e) {
                    logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
                }
                zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
                this.hookSubscribe.removeSubscribe(hook);
                dynamicTask.stop(callIdHeader.getCallId());
            }
        } else if ("push".equals(gbStream.getStreamType())) {

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/BroadcastNotifyMessageHandler.java

@@ -127,8 +127,7 @@

                // 消息发送成功， 向上级发送invite，获取推流
                try {
                    platformService.broadcastInvite(platform, deviceChannel.getChannelId(), mediaServerForMinimumLoad,  (mediaServerItem, hookParam)->{
                        OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
                    platformService.broadcastInvite(platform, deviceChannel.getChannelId(), mediaServerForMinimumLoad,  (hookData)->{
                        // 上级平台推流成功
                        AudioBroadcastCatch broadcastCatch = audioBroadcastManager.get(device.getDeviceId(), targetId);
                        if (broadcastCatch != null ) {
@@ -136,20 +135,20 @@
                                logger.info("[国标级联] 语音喊话 设备正在使用中 platform： {}， channel: {}",
                                        platform.getServerGBId(), deviceChannel.getChannelId());
                                //  查看语音通道已经建立且已经占用 回复BYE
                                platformService.stopBroadcast(platform, deviceChannel, streamChangedHookParam.getStream(),  true, mediaServerItem);
                                platformService.stopBroadcast(platform, deviceChannel, hookData.getStream(),  true, hookData.getMediaServer());
                            }else {
                                // 查看语音通道已经建立但是未占用
                                broadcastCatch.setApp(streamChangedHookParam.getApp());
                                broadcastCatch.setStream(streamChangedHookParam.getStream());
                                broadcastCatch.setMediaServerItem(mediaServerItem);
                                broadcastCatch.setApp(hookData.getApp());
                                broadcastCatch.setStream(hookData.getStream());
                                broadcastCatch.setMediaServerItem(hookData.getMediaServer());
                                audioBroadcastManager.update(broadcastCatch);
                                // 推流到设备
                                SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, targetId, streamChangedHookParam.getStream(), null);
                                SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, targetId, hookData.getStream(), null);
                                if (sendRtpItem == null) {
                                    logger.warn("[国标级联] 语音喊话 异常，未找到发流信息， channelId: {}, stream: {}", targetId, streamChangedHookParam.getStream());
                                    logger.info("[国标级联] 语音喊话 重新开始，channelId: {}, stream: {}", targetId, streamChangedHookParam.getStream());
                                    logger.warn("[国标级联] 语音喊话 异常，未找到发流信息， channelId: {}, stream: {}", targetId, hookData.getStream());
                                    logger.info("[国标级联] 语音喊话 重新开始，channelId: {}, stream: {}", targetId, hookData.getStream());
                                    try {
                                        playService.audioBroadcastCmd(device, targetId, mediaServerItem, streamChangedHookParam.getApp(), streamChangedHookParam.getStream(), 60, true, msg -> {
                                        playService.audioBroadcastCmd(device, targetId, hookData.getMediaServer(), hookData.getApp(), hookData.getStream(), 60, true, msg -> {
                                            logger.info("[语音喊话] 通道建立成功, device: {}, channel: {}", device.getDeviceId(), targetId);
                                        });
                                    } catch (SipException | InvalidArgumentException | ParseException e) {
@@ -157,7 +156,7 @@
                                    }
                                }else {
                                    // 发流
                                    JSONObject jsonObject = zlmServerFactory.startSendRtp(mediaServerItem, sendRtpItem);
                                    JSONObject jsonObject = zlmServerFactory.startSendRtp(hookData.getMediaServer(), sendRtpItem);
                                    if (jsonObject != null && jsonObject.getInteger("code") == 0 ) {
                                        logger.info("[语音喊话] 自动推流成功, device: {}, channel: {}", device.getDeviceId(), targetId);
                                    }else {
@@ -167,7 +166,7 @@
                            }
                        }else {
                            try {
                                playService.audioBroadcastCmd(device, targetId, mediaServerItem, streamChangedHookParam.getApp(), streamChangedHookParam.getStream(), 60, true, msg -> {
                                playService.audioBroadcastCmd(device, targetId, hookData.getMediaServer(), hookData.getApp(), hookData.getStream(), 60, true, msg -> {
                                    logger.info("[语音喊话] 通道建立成功, device: {}, channel: {}", device.getDeviceId(), targetId);
                                });
                            } catch (SipException | InvalidArgumentException | ParseException e) {

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java

@@ -13,9 +13,9 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.NotifyMessageHandler;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -106,9 +106,8 @@
                    logger.error("[录像流]推送完毕，收到关流通知， 发送BYE失败 {}", e.getMessage());
                }
                // 去除监听流注销自动停止下载的监听
                HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcTransaction.getStream(), false, "rtsp", ssrcTransaction.getMediaServerId());
                subscribe.removeSubscribe(hookSubscribe);

                Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcTransaction.getStream(), ssrcTransaction.getMediaServerId());
                subscribe.removeSubscribe(hook);
                // 如果级联播放，需要给上级发送此通知 TODO 多个上级同时观看一个下级 可能存在停错的问题，需要将点播CallId进行上下级绑定
                SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, ssrcTransaction.getChannelId(), null, null);
                if (sendRtpItem != null) {

 src/main/java/com/genersoft/iot/vmp/media/MediaServerConfig.java

@@ -1,7 +1,7 @@
package com.genersoft.iot.vmp.media;

import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.media.event.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import org.slf4j.Logger;

 src/main/java/com/genersoft/iot/vmp/media/abl/ABLMediaServerStatusManger.java

@@ -8,8 +8,8 @@
import com.genersoft.iot.vmp.media.abl.bean.ConfigKeyId;
import com.genersoft.iot.vmp.media.abl.event.HookAblServerKeepaliveEvent;
import com.genersoft.iot.vmp.media.abl.event.HookAblServerStartEvent;
import com.genersoft.iot.vmp.media.event.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.MediaServerDeleteEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerDeleteEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import org.slf4j.Logger;
@@ -20,11 +20,8 @@
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.io.File;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


 src/main/java/com/genersoft/iot/vmp/media/bean/MediaInfo.java

@@ -2,6 +2,7 @@

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import io.swagger.v3.oas.annotations.media.Schema;

@@ -12,6 +13,15 @@
 */
@Schema(description = "视频信息")
public class MediaInfo {
    @Schema(description = "应用名")
    private String app;
    @Schema(description = "流ID")
    private String stream;
    @Schema(description = "流媒体节点")
    private MediaServer mediaServer;
    @Schema(description = "协议")
    private String schema;

    @Schema(description = "观看人数")
    private Integer readerCount;
    @Schema(description = "视频编码类型")
@@ -37,8 +47,15 @@
    @Schema(description = "数据产生速度，单位byte/s")
    private Long bytesSpeed;

    public static MediaInfo getInstance(JSONObject jsonObject) {
    public static MediaInfo getInstance(JSONObject jsonObject, MediaServer mediaServer) {
        MediaInfo mediaInfo = new MediaInfo();
        mediaInfo.setMediaServer(mediaServer);
        String app = jsonObject.getString("app");
        mediaInfo.setApp(app);
        String stream = jsonObject.getString("stream");
        mediaInfo.setStream(stream);
        String schema = jsonObject.getString("schema");
        mediaInfo.setSchema(schema);
        Integer totalReaderCount = jsonObject.getInteger("totalReaderCount");
        Boolean online = jsonObject.getBoolean("online");
        Integer originType = jsonObject.getInteger("originType");
@@ -110,9 +127,13 @@
        return mediaInfo;
    }

    public static MediaInfo getInstance(OnStreamChangedHookParam param) {
    public static MediaInfo getInstance(OnStreamChangedHookParam param, MediaServer mediaServer) {
        List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
        MediaInfo mediaInfo = new MediaInfo();
        mediaInfo.setApp(param.getApp());
        mediaInfo.setStream(param.getStream());
        mediaInfo.setSchema(param.getSchema());
        mediaInfo.setMediaServer(mediaServer);
        mediaInfo.setReaderCount(param.getTotalReaderCount());
        mediaInfo.setOnline(param.isRegist());
        mediaInfo.setOriginType(param.getOriginType());
@@ -247,4 +268,36 @@
    public void setBytesSpeed(Long bytesSpeed) {
        this.bytesSpeed = bytesSpeed;
    }

    public String getApp() {
        return app;
    }

    public void setApp(String app) {
        this.app = app;
    }

    public String getStream() {
        return stream;
    }

    public void setStream(String stream) {
        this.stream = stream;
    }

    public MediaServer getMediaServer() {
        return mediaServer;
    }

    public void setMediaServer(MediaServer mediaServer) {
        this.mediaServer = mediaServer;
    }

    public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/bean/RecordInfo.java

New file
@@ -0,0 +1,92 @@
package com.genersoft.iot.vmp.media.bean;

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

public class RecordInfo {
    private String fileName;
    private String filePath;
    private long fileSize;
    private String folder;
    private String url;
    private long startTime;
    private double timeLen;

    public static RecordInfo getInstance(OnRecordMp4HookParam hookParam) {
        RecordInfo recordInfo = new RecordInfo();
        recordInfo.setFileName(hookParam.getFile_name());
        recordInfo.setUrl(hookParam.getUrl());
        recordInfo.setFolder(hookParam.getFolder());
        recordInfo.setFilePath(hookParam.getFile_path());
        recordInfo.setFileSize(hookParam.getFile_size());
        recordInfo.setStartTime(hookParam.getStart_time());
        recordInfo.setTimeLen(hookParam.getTime_len());
        return recordInfo;
    }

    public String getFileName() {
        return fileName;
    }

    public void setFileName(String fileName) {
        this.fileName = fileName;
    }

    public String getFilePath() {
        return filePath;
    }

    public void setFilePath(String filePath) {
        this.filePath = filePath;
    }

    public long getFileSize() {
        return fileSize;
    }

    public void setFileSize(long fileSize) {
        this.fileSize = fileSize;
    }

    public String getFolder() {
        return folder;
    }

    public void setFolder(String folder) {
        this.folder = folder;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public long getStartTime() {
        return startTime;
    }

    public void setStartTime(long startTime) {
        this.startTime = startTime;
    }

    public double getTimeLen() {
        return timeLen;
    }

    public void setTimeLen(double timeLen) {
        this.timeLen = timeLen;
    }

    @Override
    public String toString() {
        return "RecordInfo{" +
                "文件名称='" + fileName + '\'' +
                ", 文件路径='" + filePath + '\'' +
                ", 文件大小=" + fileSize +
                ", 开始时间=" + startTime +
                ", 时长=" + timeLen +
                '}';
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/MediaArrivalEvent.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/MediaDepartureEvent.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/MediaNotFoundEvent.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/MediaRtpServerTimeoutEvent.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/hook/Hook.java

New file
@@ -0,0 +1,86 @@
package com.genersoft.iot.vmp.media.event.hook;

/**
 * zlm hook事件的参数
 * @author lin
 */
public class Hook {

    private HookType hookType;

    private String app;

    private String stream;

    private String mediaServerId;

    private Long createTime;

    public static Hook getInstance(HookType hookType, String app, String stream, String mediaServerId) {
        Hook hookSubscribe = new Hook();
        hookSubscribe.setApp(app);
        hookSubscribe.setStream(stream);
        hookSubscribe.setHookType(hookType);
        hookSubscribe.setMediaServerId(mediaServerId);
        hookSubscribe.setCreateTime(System.currentTimeMillis());
        return hookSubscribe;
    }

    public HookType getHookType() {
        return hookType;
    }

    public void setHookType(HookType hookType) {
        this.hookType = hookType;
    }

    public String getApp() {
        return app;
    }

    public void setApp(String app) {
        this.app = app;
    }

    public String getStream() {
        return stream;
    }

    public void setStream(String stream) {
        this.stream = stream;
    }

    public Long getCreateTime() {
        return createTime;
    }

    public void setCreateTime(Long createTime) {
        this.createTime = createTime;
    }

    public String getMediaServerId() {
        return mediaServerId;
    }

    public void setMediaServerId(String mediaServerId) {
        this.mediaServerId = mediaServerId;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof Hook) {
            Hook param = (Hook) obj;
            return param.getHookType().equals(this.hookType)
                    && param.getApp().equals(this.app)
                    && param.getStream().equals(this.stream)
                    && param.getMediaServerId().equals(this.mediaServerId);
        }else {
            return false;
        }
    }

    @Override
    public String toString() {
        return this.getHookType() + this.getApp() + this.getStream() + this.getMediaServerId();
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookData.java

New file
@@ -0,0 +1,132 @@
package com.genersoft.iot.vmp.media.event.hook;

import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.bean.RecordInfo;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaEvent;
import com.genersoft.iot.vmp.media.event.media.MediaPublishEvent;
import com.genersoft.iot.vmp.media.event.media.MediaRecordMp4Event;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import io.swagger.v3.oas.annotations.media.Schema;

/**
 * Hook返回的内容
 */
public class HookData {
    /**
     * 应用名
     */
    private String app;
    /**
     * 流ID
     */
    private String stream;
    /**
     * 流媒体节点
     */
    private MediaServer mediaServer;
    /**
     * 协议
     */
    private String schema;

    /**
     * 流信息
     */
    private MediaInfo mediaInfo;

    /**
     * 录像信息
     */
    private RecordInfo recordInfo;

    @Schema(description = "推流的额外参数")
    private String params;
    public static HookData getInstance(MediaEvent mediaEvent) {
        HookData hookData = new HookData();
        if (mediaEvent instanceof MediaPublishEvent) {
            MediaPublishEvent event = (MediaPublishEvent) mediaEvent;
            hookData.setApp(event.getApp());
            hookData.setStream(event.getStream());
            hookData.setSchema(event.getSchema());
            hookData.setMediaServer(event.getMediaServer());
            hookData.setParams(event.getParams());
        }else if (mediaEvent instanceof MediaArrivalEvent) {
            MediaArrivalEvent event = (MediaArrivalEvent) mediaEvent;
            hookData.setApp(event.getApp());
            hookData.setStream(event.getStream());
            hookData.setSchema(event.getSchema());
            hookData.setMediaServer(event.getMediaServer());
            hookData.setMediaInfo(event.getMediaInfo());
        }else if (mediaEvent instanceof MediaRecordMp4Event) {
            MediaRecordMp4Event event = (MediaRecordMp4Event) mediaEvent;
            hookData.setApp(event.getApp());
            hookData.setStream(event.getStream());
            hookData.setSchema(event.getSchema());
            hookData.setMediaServer(event.getMediaServer());
            hookData.setRecordInfo(event.getRecordInfo());
        }else {
            hookData.setApp(mediaEvent.getApp());
            hookData.setStream(mediaEvent.getStream());
            hookData.setSchema(mediaEvent.getSchema());
            hookData.setMediaServer(mediaEvent.getMediaServer());
        }
        return hookData;
    }

    public String getApp() {
        return app;
    }

    public void setApp(String app) {
        this.app = app;
    }

    public String getStream() {
        return stream;
    }

    public void setStream(String stream) {
        this.stream = stream;
    }

    public MediaServer getMediaServer() {
        return mediaServer;
    }

    public void setMediaServer(MediaServer mediaServer) {
        this.mediaServer = mediaServer;
    }

    public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }

    public MediaInfo getMediaInfo() {
        return mediaInfo;
    }

    public void setMediaInfo(MediaInfo mediaInfo) {
        this.mediaInfo = mediaInfo;
    }

    public String getParams() {
        return params;
    }

    public void setParams(String params) {
        this.params = params;
    }

    public RecordInfo getRecordInfo() {
        return recordInfo;
    }

    public void setRecordInfo(RecordInfo recordInfo) {
        this.recordInfo = recordInfo;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookSubscribe.java

@@ -1,172 +1,105 @@
package com.genersoft.iot.vmp.media.event.hook;

import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaEvent;
import com.genersoft.iot.vmp.media.event.media.MediaPublishEvent;
import org.mybatis.logging.Logger;
import org.mybatis.logging.LoggerFactory;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.time.Instant;
import java.util.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * ZLMediaServer的hook事件订阅
 * zlm hook事件的参数
 * @author lin
 */
@Component
public class HookSubscribe {

    private final static Logger logger = LoggerFactory.getLogger(HookSubscribe.class);
    /**
     * 订阅数据过期时间
     */
    private final long subscribeExpire = 5 * 1000;

    @FunctionalInterface
    public interface Event{
        void response(MediaServer mediaServerItem, HookParam hookParam);
        void response(HookData data);
    }

    /**
     * 流到来的处理
     */
    @Async("taskExecutor")
    @org.springframework.context.event.EventListener
    @EventListener
    public void onApplicationEvent(MediaArrivalEvent event) {
        for (HookType hookType : allSubscribes.keySet()) {
            if (hookType.equals(HookType.on_stream_changed)) {

            }
        }
    }

    private Map<HookType, Map<IHookSubscribe, HookSubscribe.Event>> allSubscribes = new ConcurrentHashMap<>();

    public void addSubscribe(IHookSubscribe hookSubscribe, HookSubscribe.Event event) {
        if (hookSubscribe.getExpires() == null) {
            // 默认5分钟过期
            Instant expiresInstant = Instant.now().plusSeconds(TimeUnit.MINUTES.toSeconds(5));
            hookSubscribe.setExpires(expiresInstant);
        }
        allSubscribes.computeIfAbsent(hookSubscribe.getHookType(), k -> new ConcurrentHashMap<>()).put(hookSubscribe, event);
    }

    public HookSubscribe.Event sendNotify(HookType type, JSONObject hookResponse) {
        HookSubscribe.Event event= null;
        Map<IHookSubscribe, Event> eventMap = allSubscribes.get(type);
        if (eventMap == null) {
            return null;
        }
        for (IHookSubscribe key : eventMap.keySet()) {
            Boolean result = null;

            for (String s : key.getContent().keySet()) {
                if (result == null) {
                    result = key.getContent().getString(s).equals(hookResponse.getString(s));
                }else {
                    if (key.getContent().getString(s) == null) {
                        continue;
                    }
                    result = result && key.getContent().getString(s).equals(hookResponse.getString(s));
                }
            }
            if (null != result && result) {
                event = eventMap.get(key);
            }
        }
        return event;
    }

    public void removeSubscribe(IHookSubscribe hookSubscribe) {
        Map<IHookSubscribe, Event> eventMap = allSubscribes.get(hookSubscribe.getHookType());
        if (eventMap == null) {
            return;
        if ("rtsp".equals(event.getSchema())) {
            System.out.println("流到来的处理: " + allSubscribes.size());
            sendNotify(HookType.on_media_arrival, event);
        }

        Set<Map.Entry<IHookSubscribe, Event>> entries = eventMap.entrySet();
        if (entries.size() > 0) {
            List<Map.Entry<IHookSubscribe, HookSubscribe.Event>> entriesToRemove = new ArrayList<>();
            for (Map.Entry<IHookSubscribe, HookSubscribe.Event> entry : entries) {
                JSONObject content = entry.getKey().getContent();
                if (content == null || content.size() == 0) {
                    entriesToRemove.add(entry);
                    continue;
                }
                Boolean result = null;
                for (String s : content.keySet()) {
                    if (result == null) {
                        result = content.getString(s).equals(hookSubscribe.getContent().getString(s));
                    }else {
                        if (content.getString(s) == null) {
                            continue;
                        }
                        result = result && content.getString(s).equals(hookSubscribe.getContent().getString(s));
                    }
                }
                if (result){
                    entriesToRemove.add(entry);
                }
            }

            if (!CollectionUtils.isEmpty(entriesToRemove)) {
                for (Map.Entry<IHookSubscribe, HookSubscribe.Event> entry : entriesToRemove) {
                    eventMap.remove(entry.getKey());
                }
                if (eventMap.size() == 0) {
                    allSubscribes.remove(hookSubscribe.getHookType());
                }
            }

        }
    }

    /**
     * 获取某个类型的所有的订阅
     * @param type
     * @return
     * 流结束事件
     */
    public List<HookSubscribe.Event> getSubscribes(HookType type) {
        Map<IHookSubscribe, Event> eventMap = allSubscribes.get(type);
        if (eventMap == null) {
            return null;
    @Async("taskExecutor")
    @EventListener
    public void onApplicationEvent(MediaDepartureEvent event) {
        if ("rtsp".equals(event.getSchema())) {
            sendNotify(HookType.on_media_departure, event);
        }
        List<HookSubscribe.Event> result = new ArrayList<>();
        for (IHookSubscribe key : eventMap.keySet()) {
            result.add(eventMap.get(key));
        }
        return result;

    }
    /**
     * 推流鉴权事件
     */
    @Async("taskExecutor")
    @EventListener
    public void onApplicationEvent(MediaPublishEvent event) {
        sendNotify(HookType.on_publish, event);
    }

    public List<IHookSubscribe> getAll(){
        ArrayList<IHookSubscribe> result = new ArrayList<>();
        Collection<Map<IHookSubscribe, Event>> values = allSubscribes.values();
        for (Map<IHookSubscribe, Event> value : values) {
            result.addAll(value.keySet());
    private final Map<String, Event> allSubscribes = new ConcurrentHashMap<>();
    private final Map<String, Hook> allHook = new ConcurrentHashMap<>();

    private void sendNotify(HookType hookType, MediaEvent event) {
        Hook paramHook = Hook.getInstance(hookType, event.getApp(), event.getStream(), event.getMediaServer().getId());
        Event hookSubscribeEvent = allSubscribes.get(paramHook.toString());
        if (hookSubscribeEvent != null) {
            HookData data = HookData.getInstance(event);
            hookSubscribeEvent.response(data);
        }
        return result;
    }

    public void addSubscribe(Hook hook, HookSubscribe.Event event) {
        if (hook.getCreateTime() == null) {
            hook.setCreateTime(System.currentTimeMillis());
        }
        allSubscribes.put(hook.toString(), event);
        allHook.put(hook.toString(), hook);
    }

    public void removeSubscribe(Hook hook) {
        allSubscribes.remove(hook.toString());
        allHook.remove(hook.toString());
    }

    /**
     * 对订阅数据进行过期清理
     */
//    @Scheduled(cron="0 0/5 * * * ?")   //每5分钟执行一次
    @Scheduled(fixedRate = 2 * 1000)
    @Scheduled(fixedRate=subscribeExpire)   //每5分钟执行一次
    public void execute(){
        Instant instant = Instant.now().minusMillis(TimeUnit.MINUTES.toMillis(5));
        int total = 0;
        for (HookType hookType : allSubscribes.keySet()) {
            Map<IHookSubscribe, Event> hookSubscribeEventMap = allSubscribes.get(hookType);
            if (hookSubscribeEventMap.size() > 0) {
                for (IHookSubscribe hookSubscribe : hookSubscribeEventMap.keySet()) {
                    if (hookSubscribe.getExpires().isBefore(instant)) {
                        // 过期的
                        hookSubscribeEventMap.remove(hookSubscribe);
                        total ++;
                    }
                }
        long expireTime = System.currentTimeMillis() - subscribeExpire;
        for (Hook hook : allHook.values()) {
            if (hook.getCreateTime() < expireTime) {
                allSubscribes.remove(hook.toString());
                allHook.remove(hook.toString());
            }
        }
    }

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookSubscribeFactory.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookSubscribeForRecordMp4.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookSubscribeForRtpServerTimeout.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookSubscribeForStreamChange.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookSubscribeForStreamPush.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/hook/HookType.java

@@ -10,6 +10,6 @@
    on_publish,
    on_record_mp4,
    on_media_arrival,
    on_stream_changed,
    on_media_departure,
    on_rtp_server_timeout,
}

 src/main/java/com/genersoft/iot/vmp/media/event/hook/IHookSubscribe.java

File was deleted

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaArrivalEvent.java

New file
@@ -0,0 +1,46 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;

/**
 * 流到来事件
 */
public class MediaArrivalEvent extends MediaEvent {
    public MediaArrivalEvent(Object source) {
        super(source);
    }

    public static MediaArrivalEvent getInstance(Object source, OnStreamChangedHookParam hookParam, MediaServer mediaServer){
        MediaArrivalEvent mediaArrivalEvent = new MediaArrivalEvent(source);
        mediaArrivalEvent.setMediaInfo(MediaInfo.getInstance(hookParam, mediaServer));
        mediaArrivalEvent.setApp(hookParam.getApp());
        mediaArrivalEvent.setStream(hookParam.getStream());
        mediaArrivalEvent.setMediaServer(mediaServer);
        mediaArrivalEvent.setSchema(hookParam.getSchema());
        mediaArrivalEvent.setCallId(hookParam.getCallId());
        return mediaArrivalEvent;
    }

    private MediaInfo mediaInfo;

    private String callId;

    public MediaInfo getMediaInfo() {
        return mediaInfo;
    }

    public void setMediaInfo(MediaInfo mediaInfo) {
        this.mediaInfo = mediaInfo;
    }


    public String getCallId() {
        return callId;
    }

    public void setCallId(String callId) {
        this.callId = callId;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaDepartureEvent.java

New file
@@ -0,0 +1,22 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;

/**
 * 流离开事件
 */
public class MediaDepartureEvent extends MediaEvent {
    public MediaDepartureEvent(Object source) {
        super(source);
    }

    public static MediaDepartureEvent getInstance(Object source, OnStreamChangedHookParam hookParam, MediaServer mediaServer){
        MediaDepartureEvent mediaDepartureEven = new MediaDepartureEvent(source);
        mediaDepartureEven.setApp(hookParam.getApp());
        mediaDepartureEven.setStream(hookParam.getStream());
        mediaDepartureEven.setSchema(hookParam.getSchema());
        mediaDepartureEven.setMediaServer(mediaServer);
        return mediaDepartureEven;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaEvent.java

New file
@@ -0,0 +1,58 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import org.springframework.context.ApplicationEvent;

/**
 * 流到来事件
 */
public class MediaEvent extends ApplicationEvent {

    public MediaEvent(Object source) {
        super(source);
    }

    private String app;

    private String stream;

    private MediaServer mediaServer;

    private String schema;


    public String getApp() {
        return app;
    }

    public void setApp(String app) {
        this.app = app;
    }

    public String getStream() {
        return stream;
    }

    public void setStream(String stream) {
        this.stream = stream;
    }

    public MediaServer getMediaServer() {
        return mediaServer;
    }

    public void setMediaServer(MediaServer mediaServer) {
        this.mediaServer = mediaServer;
    }

    public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }

}

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaNotFoundEvent.java

New file
@@ -0,0 +1,22 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamNotFoundHookParam;

/**
 * 流未找到
 */
public class MediaNotFoundEvent extends MediaEvent {
    public MediaNotFoundEvent(Object source) {
        super(source);
    }

    public static MediaNotFoundEvent getInstance(Object source, OnStreamNotFoundHookParam hookParam, MediaServer mediaServer){
        MediaNotFoundEvent mediaDepartureEven = new MediaNotFoundEvent(source);
        mediaDepartureEven.setApp(hookParam.getApp());
        mediaDepartureEven.setStream(hookParam.getStream());
        mediaDepartureEven.setSchema(hookParam.getSchema());
        mediaDepartureEven.setMediaServer(mediaServer);
        return mediaDepartureEven;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaPublishEvent.java

New file
@@ -0,0 +1,33 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnPublishHookParam;

/**
 * 推流鉴权事件
 */
public class MediaPublishEvent extends MediaEvent {
    public MediaPublishEvent(Object source) {
        super(source);
    }

    public static MediaPublishEvent getInstance(Object source, OnPublishHookParam hookParam, MediaServer mediaServer){
        MediaPublishEvent mediaPublishEvent = new MediaPublishEvent(source);
        mediaPublishEvent.setApp(hookParam.getApp());
        mediaPublishEvent.setStream(hookParam.getStream());
        mediaPublishEvent.setMediaServer(mediaServer);
        mediaPublishEvent.setSchema(hookParam.getSchema());
        mediaPublishEvent.setParams(hookParam.getParams());
        return mediaPublishEvent;
    }

    private String params;

    public String getParams() {
        return params;
    }

    public void setParams(String params) {
        this.params = params;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaRecordMp4Event.java

New file
@@ -0,0 +1,36 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.bean.RecordInfo;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;

/**
 * 录像文件生成事件
 */
public class MediaRecordMp4Event extends MediaEvent {
    public MediaRecordMp4Event(Object source) {
        super(source);
    }

    private RecordInfo recordInfo;

    public static MediaRecordMp4Event getInstance(Object source, OnRecordMp4HookParam hookParam, MediaServer mediaServer){
        MediaRecordMp4Event mediaRecordMp4Event = new MediaRecordMp4Event(source);
        mediaRecordMp4Event.setApp(hookParam.getApp());
        mediaRecordMp4Event.setStream(hookParam.getStream());
        RecordInfo recordInfo = RecordInfo.getInstance(hookParam);
        mediaRecordMp4Event.setRecordInfo(recordInfo);
        mediaRecordMp4Event.setMediaServer(mediaServer);
        return mediaRecordMp4Event;
    }

    public RecordInfo getRecordInfo() {
        return recordInfo;
    }

    public void setRecordInfo(RecordInfo recordInfo) {
        this.recordInfo = recordInfo;
    }

}

 src/main/java/com/genersoft/iot/vmp/media/event/media/MediaRtpServerTimeoutEvent.java

New file
@@ -0,0 +1,22 @@
package com.genersoft.iot.vmp.media.event.media;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;

/**
 * RtpServer收流超时事件
 */
public class MediaRtpServerTimeoutEvent extends MediaEvent {
    public MediaRtpServerTimeoutEvent(Object source) {
        super(source);
    }

    public static MediaRtpServerTimeoutEvent getInstance(Object source, OnStreamChangedHookParam hookParam, MediaServer mediaServer){
        MediaRtpServerTimeoutEvent mediaDepartureEven = new MediaRtpServerTimeoutEvent(source);
        mediaDepartureEven.setApp(hookParam.getApp());
        mediaDepartureEven.setStream(hookParam.getStream());
        mediaDepartureEven.setSchema(hookParam.getSchema());
        mediaDepartureEven.setMediaServer(mediaServer);
        return mediaDepartureEven;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaSendRtpStoppedEvent.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaSendRtpStoppedEvent.java
@@ -1,4 +1,4 @@
package com.genersoft.iot.vmp.media.event;
package com.genersoft.iot.vmp.media.event.mediaServer;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamNotFoundHookParam;

 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaServerChangeEvent.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaServerChangeEvent.java
@@ -1,4 +1,4 @@
package com.genersoft.iot.vmp.media.event;
package com.genersoft.iot.vmp.media.event.mediaServer;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import org.springframework.context.ApplicationEvent;

 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaServerDeleteEvent.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaServerDeleteEvent.java
@@ -1,4 +1,4 @@
package com.genersoft.iot.vmp.media.event;
package com.genersoft.iot.vmp.media.event.mediaServer;

/**
 * zlm在线事件

 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaServerEventAbstract.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaServerEventAbstract.java
@@ -1,4 +1,4 @@
package com.genersoft.iot.vmp.media.event;
package com.genersoft.iot.vmp.media.event.mediaServer;

import org.springframework.context.ApplicationEvent;


 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaServerOfflineEvent.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaServerOfflineEvent.java
@@ -1,6 +1,4 @@
package com.genersoft.iot.vmp.media.event;

import com.genersoft.iot.vmp.media.event.MediaServerEventAbstract;
package com.genersoft.iot.vmp.media.event.mediaServer;

/**
 * zlm离线事件类

 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaServerOnlineEvent.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaServerOnlineEvent.java
@@ -1,6 +1,4 @@
package com.genersoft.iot.vmp.media.event;

import com.genersoft.iot.vmp.media.event.MediaServerEventAbstract;
package com.genersoft.iot.vmp.media.event.mediaServer;

/**
 * zlm在线事件

 src/main/java/com/genersoft/iot/vmp/media/event/mediaServer/MediaServerStatusEventListener.java

File was renamed from src/main/java/com/genersoft/iot/vmp/media/event/MediaServerStatusEventListener.java
@@ -1,4 +1,4 @@
package com.genersoft.iot.vmp.media.event;
package com.genersoft.iot.vmp.media.event.mediaServer;

import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.IStreamProxyService;

 src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java

@@ -104,4 +104,36 @@

    Map<String, String> getFFmpegCMDs(MediaServer mediaServer);

    /**
     * 根据应用名和流ID获取播放地址, 通过zlm接口检查是否存在
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId,String addr, boolean authority);


    /**
     * 根据应用名和流ID获取播放地址, 通过zlm接口检查是否存在, 返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, boolean authority);

    /**
     * 根据应用名和流ID获取播放地址, 只是地址拼接
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String callId);

    /**
     * 根据应用名和流ID获取播放地址, 只是地址拼接，返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String addr, String callId, boolean isPlay);
}

 src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java

@@ -4,17 +4,19 @@
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.MediaServerDeleteEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerDeleteEvent;
import com.genersoft.iot.vmp.media.service.IMediaNodeServerService;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
@@ -73,6 +75,10 @@

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @Autowired
    private MediaConfig mediaConfig;



    /**
@@ -714,4 +720,63 @@
        }
        return mediaNodeServerService.getFFmpegCMDs(mediaServer);
    }

    @Override
    public StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String callId) {
        return getStreamInfoByAppAndStream(mediaServerItem, app, stream, mediaInfo, null, callId, true);
    }

    @Override
    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, String addr, boolean authority) {
        StreamInfo streamInfo = null;
        if (mediaServerId == null) {
            mediaServerId = mediaConfig.getId();
        }
        MediaServer mediaInfo = getOne(mediaServerId);
        if (mediaInfo == null) {
            return null;
        }
        String calld = null;
        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
        if (streamAuthorityInfo != null) {
            calld = streamAuthorityInfo.getCallId();
        }
        List<StreamInfo> streamInfoList = getMediaList(mediaInfo, app, stream, calld);
        if (streamInfoList.isEmpty()) {
            return null;
        }else {
            return streamInfoList.get(0);
        }
    }



    @Override
    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, boolean authority) {
        return getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, null, authority);
    }

    @Override
    public StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServer, String app, String stream, MediaInfo mediaInfo, String addr, String callId, boolean isPlay) {
        StreamInfo streamInfoResult = new StreamInfo();
        streamInfoResult.setStream(stream);
        streamInfoResult.setApp(app);
        if (addr == null) {
            addr = mediaServer.getStreamIp();
        }

        streamInfoResult.setIp(addr);
        streamInfoResult.setMediaServerId(mediaServer.getId());
        String callIdParam = ObjectUtils.isEmpty(callId)?"":"?callId=" + callId;
        streamInfoResult.setRtmp(addr, mediaServer.getRtmpPort(),mediaServer.getRtmpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setRtsp(addr, mediaServer.getRtspPort(),mediaServer.getRtspSSLPort(), app,  stream, callIdParam);
        streamInfoResult.setFlv(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setFmp4(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setHls(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setTs(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setRtc(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam, isPlay);

        streamInfoResult.setMediaInfo(mediaInfo);
        return streamInfoResult;
    }
}

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

@@ -11,9 +11,10 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;

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

import com.genersoft.iot.vmp.media.bean.ResultForOnPublish;

import com.genersoft.iot.vmp.media.event.*;

import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;

import com.genersoft.iot.vmp.media.event.hook.HookType;

import com.genersoft.iot.vmp.media.event.media.*;

import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent;

import com.genersoft.iot.vmp.media.service.IMediaServerService;

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

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

@@ -175,13 +176,6 @@
            return new HookResultForOnPublish(200, "success");

        }



        taskExecutor.execute(() -> {

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

            if (subscribe != null) {

                subscribe.response(mediaServer, param);

            }

        });



        ResultForOnPublish resultForOnPublish = mediaService.authenticatePublish(mediaServer, param.getApp(), param.getStream(), param.getParams());

        if (resultForOnPublish != null) {

            HookResultForOnPublish successResult = HookResultForOnPublish.getInstance(resultForOnPublish);

@@ -316,19 +310,12 @@
            MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId());

            if (mediaServerItem != null) {

                event.setMediaServer(mediaServerItem);

                event.setApp("rtp");

                applicationEventPublisher.publishEvent(event);

            }

        }catch (Exception e) {

            logger.info("[ZLM-HOOK-rtpServer收流超时] 发送通知失败 ", e);

        }

        taskExecutor.execute(() -> {

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

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

                for (HookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

            }

        });



        return HookResult.SUCCESS();

    }

@@ -341,16 +328,16 @@
    public HookResult onRecordMp4(HttpServletRequest request, @RequestBody OnRecordMp4HookParam param) {

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



        taskExecutor.execute(() -> {

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

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

                for (HookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

        try {

            MediaRecordMp4Event event = new MediaRecordMp4Event(this);

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

            if (mediaServerItem != null) {

                event.setMediaServer(mediaServerItem);

                applicationEventPublisher.publishEvent(event);

            }

            cloudRecordService.addRecord(param);



        });

        }catch (Exception e) {

            logger.info("[ZLM-HOOK-rtpServer收流超时] 发送通知失败 ", e);

        }



        return HookResult.SUCCESS();

    }


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

@@ -158,7 +158,7 @@
                    return null;
                }
                JSONObject mediaJSON = data.getJSONObject(0);
                MediaInfo mediaInfo = MediaInfo.getInstance(mediaJSON);
                MediaInfo mediaInfo = MediaInfo.getInstance(mediaJSON, mediaServer);
                StreamInfo streamInfo = getStreamInfoByAppAndStream(mediaServer, app, stream, mediaInfo, callId, true);
                if (streamInfo != null) {
                    streamInfoList.add(streamInfo);
@@ -207,7 +207,7 @@
        if (jsonObject.getInteger("code") != 0) {
            return null;
        }
        return MediaInfo.getInstance(jsonObject);
        return MediaInfo.getInstance(jsonObject, mediaServer);
    }

    @Override

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

@@ -5,8 +5,8 @@
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.media.event.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.MediaServerDeleteEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerChangeEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerDeleteEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.ZLMServerConfig;

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamAuthorityInfo.java

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

import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;

/**
 * 流的鉴权信息

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamPushItem.java

@@ -2,7 +2,7 @@

import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.utils.DateUtil;
import io.swagger.v3.oas.annotations.media.Schema;

 src/main/java/com/genersoft/iot/vmp/service/ICloudRecordService.java

@@ -21,11 +21,6 @@
    PageInfo<CloudRecordItem> getList(int page, int count, String query,  String app, String stream, String startTime, String endTime, List<MediaServer> mediaServerItems);

    /**
     * 根据hook消息增加一条记录
     */
    void addRecord(OnRecordMp4HookParam param);

    /**
     * 获取所有的日期
     */
    List<String> getDateList(String app, String stream, int year, int month, List<MediaServer> mediaServerItems);

 src/main/java/com/genersoft/iot/vmp/service/IMediaService.java

@@ -11,39 +11,6 @@
public interface IMediaService {

    /**
     * 根据应用名和流ID获取播放地址, 通过zlm接口检查是否存在
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId,String addr, boolean authority);


    /**
     * 根据应用名和流ID获取播放地址, 通过zlm接口检查是否存在, 返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, boolean authority);

    /**
     * 根据应用名和流ID获取播放地址, 只是地址拼接
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String callId);

    /**
     * 根据应用名和流ID获取播放地址, 只是地址拼接，返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
     * @param app
     * @param stream
     * @return
     */
    StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String addr, String callId, boolean isPlay);

    /**
     * 播放鉴权
     */
    boolean authenticatePlay(String app, String stream, String callId);

 src/main/java/com/genersoft/iot/vmp/service/IPlayService.java

@@ -7,6 +7,7 @@
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
@@ -30,7 +31,7 @@
              ErrorCallback<Object> callback);
    SSRCInfo play(MediaServer mediaServerItem, String deviceId, String channelId, String ssrc, ErrorCallback<Object> callback);

    StreamInfo onPublishHandlerForPlay(MediaServer mediaServerItem, HookParam hookParam, String deviceId, String channelId);
    StreamInfo onPublishHandlerForPlay(MediaServer mediaServerItem, MediaInfo mediaInfo, String deviceId, String channelId);

    MediaServer getNewMediaServerItem(Device device);


 src/main/java/com/genersoft/iot/vmp/service/bean/CloudRecordItem.java

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.service.bean;

import com.genersoft.iot.vmp.media.event.media.MediaRecordMp4Event;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;

/**
@@ -76,18 +77,18 @@
     */
    private long timeLen;

    public static CloudRecordItem getInstance(OnRecordMp4HookParam param) {
    public static CloudRecordItem getInstance(MediaRecordMp4Event param) {
        CloudRecordItem cloudRecordItem = new CloudRecordItem();
        cloudRecordItem.setApp(param.getApp());
        cloudRecordItem.setStream(param.getStream());
        cloudRecordItem.setStartTime(param.getStart_time()*1000);
        cloudRecordItem.setFileName(param.getFile_name());
        cloudRecordItem.setFolder(param.getFolder());
        cloudRecordItem.setFileSize(param.getFile_size());
        cloudRecordItem.setFilePath(param.getFile_path());
        cloudRecordItem.setMediaServerId(param.getMediaServerId());
        cloudRecordItem.setTimeLen((long) param.getTime_len() * 1000);
        cloudRecordItem.setEndTime((param.getStart_time() + (long)param.getTime_len()) * 1000);
        cloudRecordItem.setStartTime(param.getRecordInfo().getStartTime()*1000);
        cloudRecordItem.setFileName(param.getRecordInfo().getFileName());
        cloudRecordItem.setFolder(param.getRecordInfo().getFolder());
        cloudRecordItem.setFileSize(param.getRecordInfo().getFileSize());
        cloudRecordItem.setFilePath(param.getRecordInfo().getFilePath());
        cloudRecordItem.setMediaServerId(param.getMediaServer().getId());
        cloudRecordItem.setTimeLen((long) param.getRecordInfo().getTimeLen() * 1000);
        cloudRecordItem.setEndTime((param.getRecordInfo().getStartTime() + (long)param.getRecordInfo().getTimeLen()) * 1000);
        return cloudRecordItem;
    }


 src/main/java/com/genersoft/iot/vmp/service/impl/CloudRecordServiceImpl.java

@@ -5,6 +5,8 @@
import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaRecordMp4Event;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
@@ -24,6 +26,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.time.*;
@@ -99,14 +103,15 @@
        return new ArrayList<>(resultSet);
    }

    @Override
    public void addRecord(OnRecordMp4HookParam param) {
        CloudRecordItem cloudRecordItem = CloudRecordItem.getInstance(param);
        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
    @Async("taskExecutor")
    @EventListener
    public void onApplicationEvent(MediaRecordMp4Event event) {
        CloudRecordItem cloudRecordItem = CloudRecordItem.getInstance(event);
        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(event.getApp(), event.getStream());
        if (streamAuthorityInfo != null) {
            cloudRecordItem.setCallId(streamAuthorityInfo.getCallId());
        }
        logger.info("[添加录像记录] {}/{} 文件大小：{}, 时长： {}秒", param.getApp(), param.getStream(), param.getFile_size(),param.getTime_len());
        logger.info("[添加录像记录] {}/{} 内容：{}", event.getApp(), event.getStream(), event.getRecordInfo());
        cloudRecordServiceMapper.add(cloudRecordItem);
    }


 src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java

@@ -6,8 +6,8 @@
import com.genersoft.iot.vmp.common.InviteSessionStatus;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;

 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java

@@ -50,12 +50,6 @@
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private MediaConfig mediaConfig;

    @Autowired
    private IStreamProxyService streamProxyService;

    @Autowired
@@ -87,67 +81,6 @@

    @Autowired
    private ISIPCommander commander;



    @Override
    public StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String callId) {
        return getStreamInfoByAppAndStream(mediaServerItem, app, stream, mediaInfo, null, callId, true);
    }

    @Override
    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, String addr, boolean authority) {
        StreamInfo streamInfo = null;
        if (mediaServerId == null) {
            mediaServerId = mediaConfig.getId();
        }
        MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);
        if (mediaInfo == null) {
            return null;
        }
        String calld = null;
        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
        if (streamAuthorityInfo != null) {
            calld = streamAuthorityInfo.getCallId();
        }
        List<StreamInfo> streamInfoList = mediaServerService.getMediaList(mediaInfo, app, stream, calld);
        if (streamInfoList.isEmpty()) {
            return null;
        }else {
            return streamInfoList.get(0);
        }
    }



    @Override
    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, boolean authority) {
        return getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, null, authority);
    }

    @Override
    public StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServer, String app, String stream, MediaInfo mediaInfo, String addr, String callId, boolean isPlay) {
        StreamInfo streamInfoResult = new StreamInfo();
        streamInfoResult.setStream(stream);
        streamInfoResult.setApp(app);
        if (addr == null) {
            addr = mediaServer.getStreamIp();
        }

        streamInfoResult.setIp(addr);
        streamInfoResult.setMediaServerId(mediaServer.getId());
        String callIdParam = ObjectUtils.isEmpty(callId)?"":"?callId=" + callId;
        streamInfoResult.setRtmp(addr, mediaServer.getRtmpPort(),mediaServer.getRtmpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setRtsp(addr, mediaServer.getRtspPort(),mediaServer.getRtspSSLPort(), app,  stream, callIdParam);
        streamInfoResult.setFlv(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setFmp4(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setHls(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setTs(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
        streamInfoResult.setRtc(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam, isPlay);

        streamInfoResult.setMediaInfo(mediaInfo);
        return streamInfoResult;
    }

    @Override
    public boolean authenticatePlay(String app, String stream, String callId) {

 src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java

@@ -1,9 +1,7 @@
package com.genersoft.iot.vmp.service.impl;

import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.*;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
@@ -11,10 +9,12 @@
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.MediaSendRtpStoppedEvent;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookData;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
@@ -72,7 +72,7 @@
    private IMediaServerService mediaServerService;

    @Autowired
    private SIPCommanderFroPlatform commanderForPlatform;
    private ISIPCommanderForPlatform commanderForPlatform;

    @Autowired
    private DynamicTask dynamicTask;
@@ -520,11 +520,11 @@
                    inviteStreamService.removeInviteInfo(inviteInfoForOld);
                }else {
                    // 流确实尚在推流，直接回调结果
                    OnStreamChangedHookParam hookParam = new OnStreamChangedHookParam();
                    hookParam.setApp(inviteInfoForOld.getStreamInfo().getApp());
                    hookParam.setStream(inviteInfoForOld.getStreamInfo().getStream());

                    hookEvent.response(mediaServerItemForStreamInfo, hookParam);
                    HookData hookData = new HookData();
                    hookData.setApp(inviteInfoForOld.getStreamInfo().getApp());
                    hookData.setStream(inviteInfoForOld.getStreamInfo().getStream());
                    hookData.setMediaServer(mediaServerItemForStreamInfo);
                    hookEvent.response(hookData);
                    return;
                }
            }
@@ -582,14 +582,14 @@
                }
            }
        }, userSetting.getPlayTimeout());
        commanderForPlatform.broadcastInviteCmd(platform, channelId, mediaServerItem, ssrcInfo, (mediaServerItemForInvite, hookParam)->{
        commanderForPlatform.broadcastInviteCmd(platform, channelId, mediaServerItem, ssrcInfo, (hookData)->{
            logger.info("[国标级联] 发起语音喊话 收到上级推流 deviceId: {}, channelId: {}", platform.getServerGBId(), channelId);
            dynamicTask.stop(timeOutTaskKey);
            // hook响应
            playService.onPublishHandlerForPlay(mediaServerItemForInvite, hookParam, platform.getServerGBId(), channelId);
            playService.onPublishHandlerForPlay(hookData.getMediaServer(), hookData.getMediaInfo(), platform.getServerGBId(), channelId);
            // 收到流
            if (hookEvent != null) {
                hookEvent.response(mediaServerItem, hookParam);
                hookEvent.response(hookData);
            }
        }, event -> {


 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java

@@ -17,16 +17,17 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.bean.RecordInfo;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookData;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForRecordMp4;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
@@ -98,9 +99,6 @@

    @Autowired
    private SendRtpPortManager sendRtpPortManager;

    @Autowired
    private IMediaService mediaService;

    @Autowired
    private IMediaServerService mediaServerService;
@@ -423,12 +421,12 @@

        // 查看设备是否已经在推流
        try {
            cmder.talkStreamCmd(mediaServerItem, sendRtpItem, device, channelId, callId, (mediaServerItemInuse, hookParam) -> {
                logger.info("[语音对讲] 流已生成， 开始推流： " + hookParam);
            cmder.talkStreamCmd(mediaServerItem, sendRtpItem, device, channelId, callId, (hookData) -> {
                logger.info("[语音对讲] 流已生成， 开始推流： " + hookData);
                dynamicTask.stop(timeOutTaskKey);
                // TODO 暂不做处理
            }, (mediaServerItemInuse, hookParam) -> {
                logger.info("[语音对讲] 设备开始推流： " + hookParam);
            }, (hookData) -> {
                logger.info("[语音对讲] 设备开始推流： " + hookData);
                dynamicTask.stop(timeOutTaskKey);

            }, (event) -> {
@@ -538,8 +536,7 @@
                    streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
                    mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
                    // 取消订阅消息监听
                    HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
                    subscribe.removeSubscribe(hookSubscribe);
                    subscribe.removeSubscribe(Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcInfo.getStream(), mediaServerItem.getId()));
                }
            }else {
                logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},码流：{}，端口：{}, SSRC: {}",
@@ -554,11 +551,11 @@
        }, userSetting.getPlayTimeout());

        try {
            cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channel, (mediaServerItemInuse, hookParam ) -> {
                logger.info("收到订阅消息： " + hookParam);
            cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channel, (hookData ) -> {
                logger.info("收到订阅消息： " + hookData);
                dynamicTask.stop(timeOutTaskKey);
                // hook响应
                StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInuse, hookParam, device.getDeviceId(), channel.getChannelId());
                StreamInfo streamInfo = onPublishHandlerForPlay(hookData.getMediaServer(), hookData.getMediaInfo(), device.getDeviceId(), channel.getChannelId());
                if (streamInfo == null){
                    callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
                            InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
@@ -574,7 +571,7 @@
                        streamInfo);
                logger.info("[点播成功] deviceId: {}, channelId:{}, 码流类型：{}", device.getDeviceId(), channel.getChannelId(),
                        channel.getStreamIdentification());
                snapOnPlay(mediaServerItemInuse, device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
                snapOnPlay(hookData.getMediaServer(), device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
            }, (eventResult) -> {
                // 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
                InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channel.getChannelId(),
@@ -700,11 +697,10 @@
        mediaServerService.getSnap(mediaServerItemInuse, streamUrl, 15, 1, path, fileName);
    }

    public StreamInfo onPublishHandlerForPlay(MediaServer mediaServerItem, HookParam hookParam, String deviceId, String channelId) {
    public StreamInfo onPublishHandlerForPlay(MediaServer mediaServerItem, MediaInfo mediaInfo, String deviceId, String channelId) {
        StreamInfo streamInfo = null;
        Device device = redisCatchStorage.getDevice(deviceId);
        OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
        streamInfo = onPublishHandler(mediaServerItem, streamChangedHookParam, deviceId, channelId);
        streamInfo = onPublishHandler(mediaServerItem, mediaInfo, deviceId, channelId);
        if (streamInfo != null) {
            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
            if (deviceChannel != null) {
@@ -722,9 +718,8 @@

    }

    private StreamInfo onPublishHandlerForPlayback(MediaServer mediaServerItem, HookParam param, String deviceId, String channelId, String startTime, String endTime) {
        OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam) param;
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, streamChangedHookParam, deviceId, channelId);
    private StreamInfo onPublishHandlerForPlayback(MediaServer mediaServerItem, MediaInfo mediaInfo, String deviceId, String channelId, String startTime, String endTime) {
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, mediaInfo, deviceId, channelId);
        if (streamInfo != null) {
            streamInfo.setStartTime(startTime);
            streamInfo.setEndTime(endTime);
@@ -733,7 +728,7 @@
                deviceChannel.setStreamId(streamInfo.getStream());
                storager.startPlay(deviceId, channelId, streamInfo.getStream());
            }
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(InviteSessionType.PLAYBACK, ((OnStreamChangedHookParam) param).getStream());
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(InviteSessionType.PLAYBACK, mediaInfo.getStream());
            if (inviteInfo != null) {
                inviteInfo.setStatus(InviteSessionStatus.ok);

@@ -839,10 +834,10 @@
            inviteStreamService.removeInviteInfo(inviteInfo);
        };

        HookSubscribe.Event hookEvent = (mediaServerItemInuse, hookParam) -> {
            logger.info("收到回放订阅消息： " + hookParam);
        HookSubscribe.Event hookEvent = (hookData) -> {
            logger.info("收到回放订阅消息： " + hookData);
            dynamicTask.stop(playBackTimeOutTaskKey);
            StreamInfo streamInfo = onPublishHandlerForPlayback(mediaServerItemInuse, hookParam, deviceId, channelId, startTime, endTime);
            StreamInfo streamInfo = onPublishHandlerForPlayback(hookData.getMediaServer(), hookData.getMediaInfo(), deviceId, channelId, startTime, endTime);
            if (streamInfo == null) {
                logger.warn("设备回放API调用失败！");
                callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
@@ -1028,10 +1023,10 @@
            streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
            inviteStreamService.removeInviteInfo(inviteInfo);
        };
        HookSubscribe.Event hookEvent = (mediaServerItemInuse, hookParam) -> {
            logger.info("[录像下载]收到订阅消息： " + hookParam);
        HookSubscribe.Event hookEvent = (hookData) -> {
            logger.info("[录像下载]收到订阅消息： " + hookData);
            dynamicTask.stop(downLoadTimeOutTaskKey);
            StreamInfo streamInfo = onPublishHandlerForDownload(mediaServerItemInuse, hookParam, deviceId, channelId, startTime, endTime);
            StreamInfo streamInfo = onPublishHandlerForDownload(hookData.getMediaServer(), hookData.getMediaInfo(), deviceId, channelId, startTime, endTime);
            if (streamInfo == null) {
                logger.warn("[录像下载] 获取流地址信息失败");
                callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
@@ -1049,26 +1044,24 @@
                                downLoadTimeOutTaskKey, callback, inviteInfo, InviteSessionType.DOWNLOAD);

                        // 注册录像回调事件，录像下载结束后写入下载地址
                        HookSubscribe.Event hookEventForRecord = (mediaServerItemInuse, hookParam) -> {
                        HookSubscribe.Event hookEventForRecord = (hookData) -> {
                            logger.info("[录像下载] 收到录像写入磁盘消息： ， {}/{}-{}",
                                    inviteInfo.getDeviceId(), inviteInfo.getChannelId(), ssrcInfo.getStream());
                            logger.info("[录像下载] 收到录像写入磁盘消息内容： " + hookParam);
                            OnRecordMp4HookParam recordMp4HookParam = (OnRecordMp4HookParam)hookParam;
                            String filePath = recordMp4HookParam.getFile_path();
                            logger.info("[录像下载] 收到录像写入磁盘消息内容： " + hookData);
                            RecordInfo recordInfo = hookData.getRecordInfo();
                            String filePath = recordInfo.getFilePath();
                            DownloadFileInfo downloadFileInfo = CloudRecordUtils.getDownloadFilePath(mediaServerItem, filePath);
                            InviteInfo inviteInfoForNew = inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId()
                                    , inviteInfo.getChannelId(), inviteInfo.getStream());
                            inviteInfoForNew.getStreamInfo().setDownLoadFilePath(downloadFileInfo);
                            inviteStreamService.updateInviteInfo(inviteInfoForNew);
                        };
                        HookSubscribeForRecordMp4 hookSubscribe = HookSubscribeFactory.on_record_mp4(
                                mediaServerItem.getId(), "rtp", ssrcInfo.getStream());

                        Hook hook = Hook.getInstance(HookType.on_record_mp4, "rtp", ssrcInfo.getStream(), mediaServerItem.getId());
                        // 设置过期时间，下载失败时自动处理订阅数据
//                        long difference = DateUtil.getDifference(startTime, endTime)/1000;
//                        Instant expiresInstant = Instant.now().plusSeconds(TimeUnit.MINUTES.toSeconds(difference * 2));
//                        hookSubscribe.setExpires(expiresInstant);
                        subscribe.addSubscribe(hookSubscribe, hookEventForRecord);
                        subscribe.addSubscribe(hook, hookEventForRecord);
                    });
        } catch (InvalidArgumentException | SipException | ParseException e) {
            logger.error("[命令发送失败] 录像下载: {}", e.getMessage());
@@ -1134,9 +1127,8 @@
        return inviteInfo.getStreamInfo();
    }

    private StreamInfo onPublishHandlerForDownload(MediaServer mediaServerItemInuse, HookParam hookParam, String deviceId, String channelId, String startTime, String endTime) {
        OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam) hookParam;
        StreamInfo streamInfo = onPublishHandler(mediaServerItemInuse, streamChangedHookParam, deviceId, channelId);
    private StreamInfo onPublishHandlerForDownload(MediaServer mediaServerItemInuse, MediaInfo mediaInfo, String deviceId, String channelId, String startTime, String endTime) {
        StreamInfo streamInfo = onPublishHandler(mediaServerItemInuse, mediaInfo, deviceId, channelId);
        if (streamInfo != null) {
            streamInfo.setProgress(0);
            streamInfo.setStartTime(startTime);
@@ -1153,9 +1145,8 @@
    }


    public StreamInfo onPublishHandler(MediaServer mediaServerItem, OnStreamChangedHookParam hookParam, String deviceId, String channelId) {
        MediaInfo mediaInfo = MediaInfo.getInstance(hookParam);
        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", hookParam.getStream(), mediaInfo, null);
    public StreamInfo onPublishHandler(MediaServer mediaServerItem, MediaInfo mediaInfo, String deviceId, String channelId) {
        StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", mediaInfo.getStream(), mediaInfo, null);
        streamInfo.setDeviceID(deviceId);
        streamInfo.setChannelId(channelId);
        return streamInfo;
@@ -1220,7 +1211,7 @@
        AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
        audioBroadcastResult.setApp(app);
        audioBroadcastResult.setStream(stream);
        audioBroadcastResult.setStreamInfo(new StreamContent(mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, stream, null, null, null, false)));
        audioBroadcastResult.setStreamInfo(new StreamContent(mediaServerService.getStreamInfoByAppAndStream(mediaServerItem, app, stream, null, null, null, false)));
        audioBroadcastResult.setCodec("G.711");
        return audioBroadcastResult;
    }
@@ -1591,7 +1582,7 @@
            }
        }

        talk(mediaServerItem, device, channelId, stream, (mediaServerItem1, hookParam) -> {
        talk(mediaServerItem, device, channelId, stream, (hookData) -> {
            logger.info("[语音对讲] 收到设备发来的流");
        }, eventResult -> {
            logger.warn("[语音对讲] 失败，{}/{}, 错误码 {} {}", device.getDeviceId(), channelId, eventResult.statusCode, eventResult.msg);

 src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java

@@ -9,19 +9,18 @@
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -64,9 +63,6 @@

    @Autowired
    private IVideoManagerStorage videoManagerStorager;

    @Autowired
    private IMediaService mediaService;

    @Autowired
    private ZLMServerFactory zlmServerFactory;
@@ -203,9 +199,9 @@
            callback.run(ErrorCode.ERROR100.getCode(), "保存失败", null);
            return;
        }
        HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaServer.getId());
        hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
        Hook hook = Hook.getInstance(HookType.on_media_arrival, param.getApp(), param.getStream(), mediaServer.getId());
        hookSubscribe.addSubscribe(hook, (hookData) -> {
            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
                    mediaServer, param.getApp(), param.getStream(), null, null);
            callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
        });
@@ -213,7 +209,7 @@
            String talkKey = UUID.randomUUID().toString();
            String delayTalkKey = UUID.randomUUID().toString();
            dynamicTask.startDelay(delayTalkKey, ()->{
                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaServer.getId(), false);
                StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaServer.getId(), false);
                if (streamInfo != null) {
                    callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
                }else {
@@ -223,9 +219,9 @@
            }, 7000);
            WVPResult<String> result = addStreamProxyToZlm(param);
            if (result != null && result.getCode() == 0) {
                hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
                hookSubscribe.removeSubscribe(hook);
                dynamicTask.stop(talkKey);
                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
                StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
                        mediaServer, param.getApp(), param.getStream(), null, null);
                callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
            }else {
@@ -244,7 +240,7 @@
            }
        }
        else{
            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
                    mediaServer, param.getApp(), param.getStream(), null, null);
            callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
        }

 src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java

@@ -11,8 +11,8 @@
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java

@@ -6,10 +6,10 @@
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.*;
@@ -61,9 +61,9 @@
     */
    public static final  int ERROR_CODE_TIMEOUT = -3;

    private Map<String, PlayMsgCallback> callbacks = new ConcurrentHashMap<>();
    private Map<String, PlayMsgCallbackForStartSendRtpStream> callbacksForStartSendRtpStream = new ConcurrentHashMap<>();
    private Map<String, PlayMsgErrorCallback> callbacksForError = new ConcurrentHashMap<>();
    private final Map<String, PlayMsgCallback> callbacks = new ConcurrentHashMap<>();
    private final Map<String, PlayMsgCallbackForStartSendRtpStream> callbacksForStartSendRtpStream = new ConcurrentHashMap<>();
    private final Map<String, PlayMsgErrorCallback> callbacksForError = new ConcurrentHashMap<>();

    @Autowired
    private UserSetting userSetting;
@@ -89,7 +89,7 @@
    @Autowired
    private HookSubscribe subscribe;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
    private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

    @Qualifier("taskExecutor")
    @Autowired
@@ -297,9 +297,8 @@
            }, userSetting.getPlatformPlayTimeout());

            // 添加订阅
            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtsp", mediaServerItem.getId());

            subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam)->{
            Hook hook = Hook.getInstance(HookType.on_media_arrival, content.getApp(), content.getStream(), content.getMediaServerId());
            subscribe.addSubscribe(hook, (hookData)->{
                        dynamicTask.stop(taskKey);
                        responseSendItem(mediaServerItem, content, toId, serial);
                    });

 src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -6,7 +6,7 @@
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;

 src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -9,7 +9,7 @@
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;

 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java

@@ -5,6 +5,7 @@
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.conf.security.SecurityUtils;
import com.genersoft.iot.vmp.conf.security.dto.LoginUser;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
@@ -36,9 +37,10 @@
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private IMediaService mediaService;
    @Autowired
    private IStreamProxyService streamProxyService;

    @Autowired
    private IMediaServerService mediaServerService;


    /**
@@ -85,9 +87,9 @@
            String host = request.getHeader("Host");
            String localAddr = host.split(":")[0];
            logger.info("使用{}作为返回流的ip", localAddr);
            streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
            streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
        }else {
            streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
            streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
        }

        if (streamInfo != null){
@@ -105,9 +107,9 @@
                String host = request.getHeader("Host");
                String localAddr = host.split(":")[0];
                logger.info("使用{}作为返回流的ip", localAddr);
                streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
                streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
            }else {
                streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
                streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
            }
            if (streamInfo != null){
                return new StreamContent(streamInfo);

 src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java

@@ -6,12 +6,11 @@
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
@@ -108,12 +107,11 @@
        }
        // 注册回调如果rtp收流超时则通过回调发送通知
        if (callBack != null) {
            HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId());
            Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServerItem.getId());
            // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
            hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
                    (mediaServerItemInUse, hookParam)->{
                        OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam;
                        if (stream.equals(serverTimeoutHookParam.getStream_id())) {
            hookSubscribe.addSubscribe(hook,
                    (hookData)->{
                        if (stream.equals(hookData.getStream())) {
                            logger.info("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId);
                            // 将信息写入redis中，以备后用
                            redisTemplate.delete(receiveKey);
@@ -126,7 +124,7 @@
                            } catch (IOException e) {
                                logger.error("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e);
                            }
                            hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
                            hookSubscribe.removeSubscribe(hook);
                        }
                    });
        }
@@ -241,18 +239,18 @@
        }else {
            logger.info("[第三方PS服务对接->发送流] 流不存在，等待流上线，callId->{}", callId);
            String uuid = UUID.randomUUID().toString();
            HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId());
            Hook hook = Hook.getInstance(HookType.on_media_arrival, app, stream, mediaServerItem.getId());
            dynamicTask.startDelay(uuid, ()->{
                logger.info("[第三方PS服务对接->发送流] 等待流上线超时 callId->{}", callId);
                redisTemplate.delete(key);
                hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
                hookSubscribe.removeSubscribe(hook);
            }, 10000);

            // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
            OtherPsSendInfo finalSendInfo = sendInfo;
            hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
            hookSubscribe.addSubscribe(hookSubscribeForStreamChange,
                    (mediaServerItemInUse, response)->{
            hookSubscribe.removeSubscribe(hook);
            hookSubscribe.addSubscribe(hook,
                    (hookData)->{
                        dynamicTask.stop(uuid);
                        logger.info("[第三方PS服务对接->发送流] 流上线，开始发流 callId->{}", callId);
                        try {
@@ -269,7 +267,7 @@
                            logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg"));
                            throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg"));
                        }
                        hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
                        hookSubscribe.removeSubscribe(hook);
                    });
        }
    }

 src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java

@@ -6,12 +6,11 @@
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
@@ -109,12 +108,11 @@
        }
        // 注册回调如果rtp收流超时则通过回调发送通知
        if (callBack != null) {
            HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId());
            Hook hook = Hook.getInstance(HookType.on_rtp_server_timeout, "rtp", stream, mediaServerItem.getId());
            // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
            hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
                    (mediaServerItemInUse, hookParam)->{
                        OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam;
                        if (stream.equals(serverTimeoutHookParam.getStream_id())) {
            hookSubscribe.addSubscribe(hook,
                    (hookData)->{
                        if (stream.equals(hookData.getStream())) {
                            logger.info("[开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId);
                            OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
                            OkHttpClient client = httpClientBuilder.build();
@@ -125,7 +123,7 @@
                            } catch (IOException e) {
                                logger.error("[第三方服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e);
                            }
                            hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
                            hookSubscribe.removeSubscribe(hook);
                        }
                    });
        }
@@ -225,7 +223,7 @@
        if (!((dstPortForAudio > 0 && !ObjectUtils.isEmpty(dstPortForAudio) || (dstPortForVideo > 0 && !ObjectUtils.isEmpty(dstIpForVideo))))) {
            throw new ControllerException(ErrorCode.ERROR400.getCode(), "至少应该存在一组音频或视频发送参数");
        }
        MediaServer mediaServerItem = mediaServerService.getDefaultMediaServer();
        MediaServer mediaServer = mediaServerService.getDefaultMediaServer();
        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
        OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
        if (sendInfo == null) {
@@ -278,10 +276,10 @@
            paramForVideo = null;
        }

        Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream);
        Boolean streamReady = zlmServerFactory.isStreamReady(mediaServer, app, stream);
        if (streamReady) {
            if (paramForVideo != null) {
                JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForVideo);
                JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForVideo);
                if (jsonObject.getInteger("code") == 0) {
                    logger.info("[第三方服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, paramForVideo);
                    redisTemplate.opsForValue().set(key, sendInfo);
@@ -292,7 +290,7 @@
                }
            }
            if(paramForAudio != null) {
                JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForAudio);
                JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForAudio);
                if (jsonObject.getInteger("code") == 0) {
                    logger.info("[第三方服务对接->发送流] 音频流发流成功，callId->{}，param->{}", callId, paramForAudio);
                    redisTemplate.opsForValue().set(key, sendInfo);
@@ -305,18 +303,18 @@
        }else {
            logger.info("[第三方服务对接->发送流] 流不存在，等待流上线，callId->{}", callId);
            String uuid = UUID.randomUUID().toString();
            HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId());
            Hook hook = Hook.getInstance(HookType.on_media_arrival, app, stream, mediaServer.getId());
            dynamicTask.startDelay(uuid, ()->{
                logger.info("[第三方服务对接->发送流] 等待流上线超时 callId->{}", callId);
                redisTemplate.delete(key);
                hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
                hookSubscribe.removeSubscribe(hook);
            }, 10000);

            // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
            OtherRtpSendInfo finalSendInfo = sendInfo;
            hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
            hookSubscribe.addSubscribe(hookSubscribeForStreamChange,
                    (mediaServerItemInUse, response)->{
            hookSubscribe.removeSubscribe(hook);
            hookSubscribe.addSubscribe(hook,
                    (hookData)->{
                        dynamicTask.stop(uuid);
                        logger.info("[第三方服务对接->发送流] 流上线，开始发流 callId->{}", callId);
                        try {
@@ -325,7 +323,7 @@
                            throw new RuntimeException(e);
                        }
                        if (paramForVideo != null) {
                            JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForVideo);
                            JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForVideo);
                            if (jsonObject.getInteger("code") == 0) {
                                logger.info("[第三方服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, paramForVideo);
                                redisTemplate.opsForValue().set(key, finalSendInfo);
@@ -336,7 +334,7 @@
                            }
                        }
                        if(paramForAudio != null) {
                            JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, paramForAudio);
                            JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServer, paramForAudio);
                            if (jsonObject.getInteger("code") == 0) {
                                logger.info("[第三方服务对接->发送流] 音频流发流成功，callId->{}，param->{}", callId, paramForAudio);
                                redisTemplate.opsForValue().set(key, finalSendInfo);
@@ -346,7 +344,7 @@
                                throw new ControllerException(ErrorCode.ERROR100.getCode(), "[音频流发流失败] " + jsonObject.getString("msg"));
                            }
                        }
                        hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
                        hookSubscribe.removeSubscribe(hook);
                    });
        }
    }

 src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java

@@ -9,10 +9,9 @@
import com.genersoft.iot.vmp.conf.VersionInfo;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.IHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
@@ -40,8 +39,6 @@
@RequestMapping("/api/server")
public class ServerController {

    @Autowired
    private HookSubscribe zlmHttpHookSubscribe;

    @Autowired
    private IMediaServerService mediaServerService;
@@ -76,8 +73,6 @@
    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private SendRtpPortManager sendRtpPortManager;


    @GetMapping(value = "/media_server/list")
@@ -215,13 +210,6 @@
            }
        }
        return jsonObject;
    }

    @GetMapping(value = "/hooks")
    @ResponseBody
    @Operation(summary = "获取当前所有hook")
    public List<IHookSubscribe> getHooks() {
        return zlmHttpHookSubscribe.getAll();
    }

    @GetMapping(value = "/system/info")

 src/main/java/com/genersoft/iot/vmp/vmanager/streamPush/StreamPushController.java

@@ -249,7 +249,7 @@
        if (push != null && !push.isSelf()) {
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "来自其他平台的推流信息");
        }
        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
        StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
        if (streamInfo == null){
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取播放地址失败");
        }