优化媒体节点服务的代码结构

648540858

2024-03-23 c1ac16bf620d09a94a0cfe184cbe1c9cb531a01b

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

@@ -10,20 +10,20 @@
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;

import com.genersoft.iot.vmp.gb28181.bean.*;

import com.genersoft.iot.vmp.gb28181.event.EventPublisher;

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

import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerKeepaliveEvent;

import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerStartEvent;

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

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

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

@@ -38,6 +38,7 @@
import org.slf4j.LoggerFactory;

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

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

import org.springframework.context.ApplicationEventPublisher;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.util.ObjectUtils;

@@ -72,9 +73,6 @@


    @Autowired

    private AudioBroadcastManager audioBroadcastManager;



    @Autowired

    private ZLMServerFactory zlmServerFactory;



    @Autowired

    private IPlayService playService;

@@ -125,9 +123,6 @@
    private VideoStreamSessionManager sessionManager;



    @Autowired

    private AssistRESTfulUtils assistRESTfulUtils;



    @Autowired

    private SSRCFactory ssrcFactory;



    @Qualifier("taskExecutor")

@@ -137,6 +132,9 @@
    @Autowired

    private RedisTemplate<Object, Object> redisTemplate;



    @Autowired

    private ApplicationEventPublisher applicationEventPublisher;



    /**

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

     */

@@ -144,18 +142,16 @@


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

    public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {





        taskExecutor.execute(() -> {

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

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

        try {

            HookZlmServerKeepaliveEvent event = new HookZlmServerKeepaliveEvent(this);

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

            if (mediaServerItem != null) {

                event.setMediaServerItem(mediaServerItem);

                applicationEventPublisher.publishEvent(event);

            }

        });

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



        }catch (Exception e) {

            logger.info("[ZLM-HOOK-心跳] 发送通知失败 ", e);

        }

        return HookResult.SUCCESS();

    }



@@ -163,11 +159,10 @@
     * 播放器鉴权事件，rtsp/rtmp/http-flv/ws-flv/hls的播放都将触发此鉴权事件。

     */

    @ResponseBody



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

    public HookResult onPlay(@RequestBody OnPlayHookParam param) {

        if (logger.isDebugEnabled()) {

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

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

        }

        String mediaServerId = param.getMediaServerId();



@@ -175,12 +170,13 @@
            JSONObject json = (JSONObject) JSON.toJSON(param);

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

            if (subscribe != null) {

                MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);

                MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);

                if (mediaInfo != null) {

                    subscribe.response(mediaInfo, param);

                }

            }

        });

        // TODO 此处逻辑适合迁移到MediaService中

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

            Map<String, String> paramMap = urlParamToMap(param.getParams());

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

@@ -202,10 +198,11 @@
        JSONObject json = (JSONObject) JSON.toJSON(param);



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

        // TODO 加快处理速度



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

        MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);

        if (mediaInfo == null) {

        MediaServer mediaServer = mediaServerService.getOne(mediaServerId);

        if (mediaServer == null) {

            return new HookResultForOnPublish(200, "success");

        }

        // 推流鉴权的处理

@@ -253,11 +250,7 @@
        taskExecutor.execute(() -> {

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

            if (subscribe != null) {

                if (mediaInfo != null) {

                    subscribe.response(mediaInfo, param);

                } else {

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

                }

                subscribe.response(mediaServer, param);

            }

        });



@@ -273,7 +266,7 @@
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());



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

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

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

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

                inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);

                if (inviteInfo != null) {

@@ -355,7 +348,7 @@
        JSONObject json = (JSONObject) JSON.toJSON(param);

        taskExecutor.execute(() -> {

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

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

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

            if (mediaInfo == null) {

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

                return;

@@ -364,7 +357,6 @@
                subscribe.response(mediaInfo, param);

            }



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

            // TODO 重构此处逻辑

            if (param.isRegist()) {

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

@@ -468,7 +460,7 @@
                                callId = streamAuthorityInfo.getCallId();

                            }

                            StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaInfo,

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

                                    param.getApp(), param.getStream(), MediaInfo.getInstance(param), callId);

                            param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));

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

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

@@ -580,9 +572,9 @@
                }

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

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

                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(

                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(

                            inviteInfo.getChannelId());

                    if (sendRtpItems.size() > 0) {

                    if (!sendRtpItems.isEmpty()) {

                        for (SendRtpItem sendRtpItem : sendRtpItems) {

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

                            try {

@@ -676,7 +668,7 @@


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



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

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

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

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

            return defaultResult;

@@ -799,13 +791,22 @@
        logger.info("[ZLM HOOK] zlm 启动 " + zlmServerConfig.getGeneralMediaServerId());

        taskExecutor.execute(() -> {

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

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

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, zlmServerConfig);

                }

            }

            mediaServerService.zlmServerOnline(zlmServerConfig);

        });

        try {

            HookZlmServerStartEvent event = new HookZlmServerStartEvent(this);

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

            if (mediaServerItem != null) {

                event.setMediaServerItem(mediaServerItem);

                applicationEventPublisher.publishEvent(event);

            }

        }catch (Exception e) {

            logger.info("[ZLM-HOOK-ZLM启动] 发送通知失败 ", e);

        }



        return HookResult.SUCCESS();

    }

@@ -848,12 +849,11 @@
     */

    @ResponseBody

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

    public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam

    public HookResult onRtpServerTimeout(@RequestBody OnRtpServerTimeoutHookParam

            param) {

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



        taskExecutor.execute(() -> {

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

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

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {