支持对接第三方PS流

648540858

2023-08-08 5e34039deffa57fc50e93e45b6fa6879ce70e552

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

@@ -5,6 +5,7 @@
import com.genersoft.iot.vmp.common.InviteInfo;

import com.genersoft.iot.vmp.common.InviteSessionType;

import com.genersoft.iot.vmp.common.StreamInfo;

import com.genersoft.iot.vmp.common.VideoManagerConstants;

import com.genersoft.iot.vmp.conf.UserSetting;

import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;

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

@@ -22,14 +23,20 @@
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;

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

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

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

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

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

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

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

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

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

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

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

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

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

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

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

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.util.ObjectUtils;

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

@@ -116,6 +123,9 @@
    @Autowired

    private ThreadPoolTaskExecutor taskExecutor;



    @Autowired

    private RedisTemplate<Object, Object> redisTemplate;



    /**

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

     */

@@ -124,14 +134,12 @@
    @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")

    public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {



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



        taskExecutor.execute(() -> {

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

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

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, json);

                    subscribe.response(null, param);

                }

            }

        });

@@ -158,7 +166,7 @@
            if (subscribe != null) {

                MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);

                if (mediaInfo != null) {

                    subscribe.response(mediaInfo, json);

                    subscribe.response(mediaInfo, param);

                }

            }

        });

@@ -186,7 +194,10 @@


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

        MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);



        if (mediaInfo == null) {

            return new HookResultForOnPublish(200, "success");

        }

        // 推流鉴权的处理

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

            if (userSetting.getPushAuthority()) {

                // 推流鉴权

@@ -226,25 +237,32 @@




        HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();

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

            result.setEnable_audio(true);

        }



        result.setEnable_audio(true);

        taskExecutor.execute(() -> {

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

            if (subscribe != null) {

                if (mediaInfo != null) {

                    subscribe.response(mediaInfo, json);

                    subscribe.response(mediaInfo, param);

                } else {

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

                }

            }

        });



        // 是否录像

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

            result.setEnable_mp4(userSetting.getRecordSip());

        } else {

            result.setEnable_mp4(userSetting.isRecordPushLive());

        }

        // 替换流地址

        if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) {

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

            InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);

            if (inviteInfo != null) {

                result.setStream_replace(inviteInfo.getStream());

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

            }

        }

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

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

@@ -257,7 +275,6 @@
            // 如果是录像下载就设置视频间隔十秒

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

                result.setMp4_max_second(10);

                result.setEnable_audio(true);

                result.setEnable_mp4(true);

            }

        }

@@ -279,6 +296,17 @@
                }

            }

        }

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

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

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



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

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

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

                result.setEnable_mp4(true);

            }

        }

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

        return result;

    }



@@ -306,7 +334,7 @@
                return;

            }

            if (subscribe != null) {

                subscribe.response(mediaInfo, json);

                subscribe.response(mediaInfo, param);

            }



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

@@ -345,19 +373,10 @@
                }



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

                    if(param.getStream().split("_").length == 3){

                        boolean isSubStream = "sub".equals(param.getStream().split("_")[0]);

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

                        if(inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY )){

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

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

                        }

                    }else {

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

                        if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) {

                            inviteStreamService.removeInviteInfo(inviteInfo);

                            storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());

                        }

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

                    if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) {

                        inviteStreamService.removeInviteInfo(inviteInfo);

                        storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());

                    }

                } else {

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

@@ -395,7 +414,9 @@
                        }

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

                        if (gbStream != null) {

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

                            if (userSetting.isUsePushingAsStatus()) {

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

                            }

                        }

                        if (type != null) {

                            // 发送流变化redis消息

@@ -455,8 +476,6 @@
        if ("rtp".equals(param.getApp())) {

            ret.put("close", userSetting.getStreamOnDemand());

            // 国标流， 点播/录像回放/录像下载

//            StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());



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

            // 点播

            if (inviteInfo != null) {

@@ -479,21 +498,22 @@
                            }

                            redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),

                                    sendRtpItem.getCallId(), sendRtpItem.getStreamId());

                            if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {

                                MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,

                                        sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),

                                        sendRtpItem.getPlatformId(), parentPlatform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());

                                messageForPushChannel.setPlatFormIndex(parentPlatform.getId());

                                redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);

                            }

                        }

                    }

                }

                Device device = deviceService.getDevice(inviteInfo.getDeviceId());

                if (device != null) {

                    try {

                        InviteInfo info = null;

                        if(device.isSwitchPrimarySubStream()){

                            boolean isSubStream = "sub".equals(param.getStream().split("_")[0]);

                            info = inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(),isSubStream, inviteInfo.getStream());

                        }else {

                            info = inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());



                        }



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

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

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

                        if (info != null) {

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

                                    inviteInfo.getStream(), null);

@@ -504,15 +524,9 @@
                    }

                }



                if(device.isSwitchPrimarySubStream()){

                    boolean isSubStream = "sub".equals(param.getStream().split("_")[0]);

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

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

                }else {

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

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

                    storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());

                }

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

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

                storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());

                return ret;

            }

        } else {

@@ -520,7 +534,7 @@
            // 拉流代理

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

            if (streamProxyItem != null) {

                if (streamProxyItem.isEnableDisableNoneReader()) {

                if (streamProxyItem.isEnableRemoveNoneReader()) {

                    // 无人观看自动移除

                    ret.put("close", true);

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

@@ -537,7 +551,7 @@
                }

                return ret;

            }

            // 推流具有主动性，暂时不做处理

            // TODO 推流具有主动性，暂时不做处理

//         StreamPushItem streamPushItem = streamPushService.getPush(app, streamId);

//         if (streamPushItem != null) {

//            // TODO 发送停止

@@ -565,26 +579,12 @@


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

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

            if (!mediaInfo.isRtpEnable() ) {

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }else if(s.length != 2 && s.length != 3 ){

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

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }

            String deviceId = null;

            String channelId = null;

            boolean isSubStream = false;

            if (s[0].length() < 20) {

                if ("sub".equals(s[0])) {

                    isSubStream = true;

                }

                deviceId = s[1];

                channelId = s[2];

            } else {

                deviceId = s[0];

                channelId = s[1];

            }

            String deviceId = s[0];

            String channelId = s[1];

            Device device = redisCatchStorage.getDevice(deviceId);

            if (device == null) {

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

@@ -595,33 +595,78 @@
                defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));

                return defaultResult;

            }

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

            if (s.length == 2) {

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



            RequestMessage msg = new RequestMessage();

            String key = DeferredResultHolder.getPlayKey(deviceId, channelId, device.isSwitchPrimarySubStream(), isSubStream);

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

            msg.setKey(key);

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

            msg.setId(uuid);

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

                RequestMessage msg = new RequestMessage();

                String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;

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

                msg.setKey(key);

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

                msg.setId(uuid);

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



            result.onTimeout(() -> {

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

                // 释放rtpserver

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

                resultHolder.invokeResult(msg);

            });



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

            resultHolder.put(key, uuid, result);



            if (!exist) {

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

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

                result.onTimeout(() -> {

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

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

                    resultHolder.invokeResult(msg);

                });



                resultHolder.put(key, uuid, result);



                if (!exist) {

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

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

                        resultHolder.invokeResult(msg);

                    });

                }

                return result;

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

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

                String startTimeStr = s[2];

                String endTimeStr = s[3];

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

                    defaultResult.setResult(HookResult.SUCCESS());

                    return defaultResult;

                }

                String startTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(startTimeStr);

                String endTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(endTimeStr);

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

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

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

                        startTime, endTime

                );

                RequestMessage msg = new RequestMessage();

                String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;

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

                msg.setKey(key);

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

                msg.setId(uuid);

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



                result.onTimeout(() -> {

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

                    // 释放rtpserver

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

                    resultHolder.invokeResult(msg);

                });



                resultHolder.put(key, uuid, result);



                if (!exist) {

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

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

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

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

                        resultHolder.invokeResult(msg);

                    });

                }

                return result;

            }else {

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }

            return result;



        } else {

            // 拉流代理

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

@@ -649,7 +694,7 @@
            List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_started);

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, jsonObject);

                    subscribe.response(null, zlmServerConfig);

                }

            }

            mediaServerService.zlmServerOnline(zlmServerConfig);

@@ -704,7 +749,7 @@
            List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_rtp_server_timeout);

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, json);

                    subscribe.response(null, param);

                }

            }

        });