修复目录订阅发送订阅未携带经纬度的问题

648540858

2024-04-15 bd0fafde817b60dee79b1e4dd091a5e34ac8625c

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

@@ -73,9 +73,6 @@
    private AudioBroadcastManager audioBroadcastManager;



    @Autowired

    private ZLMServerFactory zlmServerFactory;



    @Autowired

    private IPlayService playService;



    @Autowired

@@ -124,9 +121,6 @@
    private VideoStreamSessionManager sessionManager;



    @Autowired

    private AssistRESTfulUtils assistRESTfulUtils;



    @Autowired

    private SSRCFactory ssrcFactory;



    @Qualifier("taskExecutor")

@@ -147,7 +141,7 @@


        taskExecutor.execute(() -> {

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

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

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

                for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

                    subscribe.response(null, param);

                }

@@ -166,7 +160,7 @@
    @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();



@@ -252,11 +246,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(mediaInfo, param);

            }

        });



@@ -270,7 +260,6 @@
        if ("rtp".equals(param.getApp())) {



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



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

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

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

@@ -278,6 +267,8 @@
                if (inviteInfo != null) {

                    result.setStream_replace(inviteInfo.getStream());

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

                    // 单端口模式下修改流ID为目标流ID，不然其他地方可能都无法对应

                    param.setStream(inviteInfo.getStream());

                }

            }



@@ -519,32 +510,46 @@
                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());

                    if (!sendRtpItems.isEmpty()) {

                        for (SendRtpItem sendRtpItem : sendRtpItems) {

                            if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) {

                                String platformId = sendRtpItem.getPlatformId();

                                ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);

                                Device device = deviceService.getDevice(platformId);

                            if (sendRtpItem == null) {

                                continue;

                            }



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

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

                                                }

                                            }

                                        }

                                    } catch (SipException | InvalidArgumentException | ParseException |

                                             SsrcTransactionNotFoundException e) {

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

                                    }

                                } catch (SipException | InvalidArgumentException | ParseException |

                                         SsrcTransactionNotFoundException e) {

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

                                }



                            }

                        }

                    }

@@ -798,7 +803,7 @@
        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);

                }

@@ -847,12 +852,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) {