支持彻底关闭文档页面 + 修复云端录像偶现callId错误问题

648540858

2024-05-23 2daa59d78ce206308c35294da04cffe5e340143a

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

@@ -10,22 +10,26 @@
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.zlm.dto.*;

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.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.service.redisMsg.IRedisRpcService;

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.utils.MediaServerUtils;

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

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

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

@@ -44,7 +48,6 @@
import javax.sip.InvalidArgumentException;

import javax.sip.SipException;

import java.text.ParseException;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.UUID;

@@ -77,6 +80,10 @@


    @Autowired

    private IRedisCatchStorage redisCatchStorage;





    @Autowired

    private IRedisRpcService redisRpcService;



    @Autowired

    private IInviteStreamService inviteStreamService;

@@ -172,7 +179,7 @@
            }

        });

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

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

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

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

            if (streamAuthorityInfo != null && streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(paramMap.get("callId"))) {

                return new HookResult(401, "Unauthorized");

@@ -213,7 +220,7 @@
                    logger.info("推流鉴权失败： 缺少必要参数：sign=md5(user表的pushKey)");

                    return new HookResultForOnPublish(401, "Unauthorized");

                }

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

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

                String sign = paramMap.get("sign");

                if (sign == null) {

                    logger.info("推流鉴权失败： 缺少必要参数：sign=md5(user表的pushKey)");

@@ -282,7 +289,7 @@
                String channelId = ssrcTransactionForAll.get(0).getChannelId();

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

                if (deviceChannel != null) {

                    result.setEnable_audio(deviceChannel.isHasAudio());

                    result.setEnable_audio(deviceChannel.getHasAudio());

                }

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

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

@@ -342,6 +349,11 @@
            MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());

            if (mediaInfo == null) {

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

                return;

            }

            if (!ObjectUtils.isEmpty(mediaInfo.getTranscodeSuffix())

                    && !"null".equalsIgnoreCase(mediaInfo.getTranscodeSuffix())

                    && param.getStream().endsWith(mediaInfo.getTranscodeSuffix())  ) {

                return;

            }

            if (subscribe != null) {

@@ -509,43 +521,35 @@
                            if (sendRtpItem == null) {

                                continue;

                            }



                            if (sendRtpItem.getApp().equals(param.getApp())) {

                                logger.info(sendRtpItem.toString());

                                if (userSetting.getServerId().equals(sendRtpItem.getServerId())) {

                                    MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,

                                            sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),

                                            sendRtpItem.getPlatformId(), null, userSetting.getServerId(), param.getMediaServerId());

                                    // 通知其他wvp停止发流

                                    redisCatchStorage.sendPushStreamClose(messageForPushChannel);

                                }else {

                                    String platformId = sendRtpItem.getPlatformId();

                                    ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);

                                    Device device = deviceService.getDevice(platformId);



                                    try {

                                        if (platform != null) {

                                            commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

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

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

                                        } else {

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

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

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

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

                                                if (audioBroadcastCatch != null) {

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

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

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

                                                }

                                // 在hook收到这个消息，说明发流一定是本级完成的。

                                ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());

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

                                Device device = deviceService.getDevice(sendRtpItem.getPlatformId());

                                try {

                                    if (platform != null) {

                                        commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                                        redisCatchStorage.deleteSendRTPServer(sendRtpItem);

                                        redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, platform);

                                    } else if (device != null) {

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

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

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

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

                                            if (audioBroadcastCatch != null) {

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

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

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

                                            }

                                        }

                                    } catch (SipException | InvalidArgumentException | ParseException |

                                             SsrcTransactionNotFoundException e) {

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

                                    }else {

                                        // 通知其他wvp停止发流

                                        redisRpcService.rtpSendStopped(sendRtpItem.getRedisKey());

                                    }

                                } catch (SipException | InvalidArgumentException | ParseException |

                                         SsrcTransactionNotFoundException e) {

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

                                }



                            }

                        }

                    }

@@ -564,6 +568,19 @@


        logger.info("[ZLM HOOK]流无人观看：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(),

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



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

        if (mediaInfo == null) {

            JSONObject ret = new JSONObject();

            ret.put("code", 0);

            return ret;

        }

        if (!ObjectUtils.isEmpty(mediaInfo.getTranscodeSuffix())

                && !"null".equalsIgnoreCase(mediaInfo.getTranscodeSuffix())

                && param.getStream().endsWith(mediaInfo.getTranscodeSuffix())  ) {

            param.setStream(param.getStream().substring(0, param.getStream().lastIndexOf(mediaInfo.getTranscodeSuffix()) -1 ));

        }



        JSONObject ret = new JSONObject();

        ret.put("code", 0);

        // 国标类型的流

@@ -590,14 +607,10 @@
                            } catch (SipException | InvalidArgumentException | ParseException e) {

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

                            }

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

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

                            redisCatchStorage.deleteSendRTPServer(sendRtpItem);

                            ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());

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

                                MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,

                                        sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),

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

                                messageForPushChannel.setPlatFormIndex(parentPlatform.getId());

                                redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);

                                redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, parentPlatform);

                            }

                        }

                    }

@@ -825,17 +838,18 @@
        }

        taskExecutor.execute(() -> {

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

            if (sendRtpItems.size() > 0) {

            if (!sendRtpItems.isEmpty()) {

                for (SendRtpItem sendRtpItem : sendRtpItems) {

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

                    ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());

                    try {

                        commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());

                    } catch (SipException | InvalidArgumentException | ParseException e) {

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

                    if(parentPlatform != null) {

                        try {

                            commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());

                        } catch (SipException | InvalidArgumentException | ParseException e) {

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

                        }

                    }

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

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

                    ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());

                    redisCatchStorage.deleteSendRTPServer(sendRtpItem);

                }

            }

        });

@@ -884,23 +898,5 @@
        });



        return HookResult.SUCCESS();

    }



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

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

        if (ObjectUtils.isEmpty(params)) {

            return map;

        }

        String[] paramsArray = params.split("&");

        if (paramsArray.length == 0) {

            return map;

        }

        for (String param : paramsArray) {

            String[] paramArray = param.split("=");

            if (paramArray.length == 2) {

                map.put(paramArray[0], paramArray[1]);

            }

        }

        return map;

    }

}