Merge pull request #893 from sxh-netizen/wvp-28181-2.0

648540858

2023-06-20 fa62ab9a0143433a5d058ab3229a37e4a9a0f696

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

@@ -289,6 +289,7 @@
    @ResponseBody

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

    public HookResult onStreamChanged(@RequestBody OnStreamChangedHookParam param) {



        if (param.isRegist()) {

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

        } else {

@@ -310,11 +311,13 @@


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

            // TODO 重构此处逻辑

            boolean isPush = false;

            if (param.isRegist()) {

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

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

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

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

                    isPush = true;

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

                    if (streamAuthorityInfo == null) {

                        streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);

@@ -328,7 +331,7 @@
                redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());

            }



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

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

                // 更新流媒体负载信息

                if (param.isRegist()) {

                    mediaServerService.addCount(param.getMediaServerId());

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



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

                        }

                    }

                } else {

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

@@ -360,8 +372,6 @@
                            StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaInfo,

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

                            param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));

                            // 如果是拉流代理产生的，不需要写入推流



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

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

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

@@ -450,6 +460,11 @@
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());

            // 点播

            if (inviteInfo != null) {

                // 录像下载

                if (inviteInfo.getType() == InviteSessionType.DOWNLOAD) {

                    ret.put("close", false);

                    return ret;

                }

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

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

                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(

@@ -467,18 +482,19 @@
                        }

                    }

                }



                if (userSetting.getStreamOnDemand()) {

                    // 录像下载

                    if (inviteInfo.getType() == InviteSessionType.DOWNLOAD) {

                        ret.put("close", false);

                        return ret;

                    }



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

                    if (device != null) {

                        try {

                            if (inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()) != null) {

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



                        }



                        if (info != null) {

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

                                        inviteInfo.getStream(), null);

                            }

@@ -488,6 +504,11 @@
                        }

                    }



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

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

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

            if (streamProxyItem != null) {

                if (streamProxyItem.isEnableRemoveNoneReader()) {

                if (streamProxyItem.isEnableDisableNoneReader()) {

                    // 无人观看自动移除

                    ret.put("close", true);

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

@@ -544,12 +565,26 @@


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

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

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

            if (!mediaInfo.isRtpEnable() ) {

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

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

                defaultResult.setResult(HookResult.SUCCESS());

                return defaultResult;

            }

            String deviceId = s[0];

            String channelId = s[1];

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

            }

            Device device = redisCatchStorage.getDevice(deviceId);

            if (device == null) {

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

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



            RequestMessage msg = new RequestMessage();

            String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;

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

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

            msg.setKey(key);

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

@@ -581,7 +616,7 @@
            resultHolder.put(key, uuid, result);



            if (!exist) {

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

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

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

                    resultHolder.invokeResult(msg);

                });