优化国标级联发流并发能力

648540858

2022-11-16 694076dc8c447754b02dff24a891249f54427ffb

优化国标级联发流并发能力

 sql/mysql.sql

@@ -281,7 +281,6 @@
                                `secret` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                                `rtpEnable` int NOT NULL,
                                `rtpPortRange` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                                `sendRtpPortRange` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
                                `recordAssistPort` int NOT NULL,
                                `defaultServer` int NOT NULL,
                                `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,

 sql/update.sql

@@ -1,6 +1,9 @@
alter table media_server
    drop column streamNoneReaderDelayMS;

alter table media_server
    drop column sendRtpPortRange;

alter table stream_proxy
    add enable_disable_none_reader bit(1) default null;


 src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java

@@ -37,6 +37,8 @@

    private Boolean pushAuthority = Boolean.TRUE;

    private Boolean gbSendStreamStrict = Boolean.FALSE;

    private String serverId = "000000";

    private String thirdPartyGBIdReg = "[\\s\\S]*";
@@ -166,4 +168,12 @@
    public void setPushAuthority(Boolean pushAuthority) {
        this.pushAuthority = pushAuthority;
    }

    public Boolean getGbSendStreamStrict() {
        return gbSendStreamStrict;
    }

    public void setGbSendStreamStrict(Boolean gbSendStreamStrict) {
        this.gbSendStreamStrict = gbSendStreamStrict;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java

@@ -7,10 +7,12 @@
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.sip.*;
import javax.sip.DialogTerminatedEvent;
import javax.sip.ResponseEvent;
import javax.sip.TimeoutEvent;
import javax.sip.TransactionTerminatedEvent;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.time.Instant;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -29,6 +31,7 @@
    private Map<String, SipSubscribe.Event> okSubscribes = new ConcurrentHashMap<>();

    private Map<String, Instant> okTimeSubscribes = new ConcurrentHashMap<>();

    private Map<String, Instant> errorTimeSubscribes = new ConcurrentHashMap<>();

    //    @Scheduled(cron="*/5 * * * * ?")   //每五秒执行一次

 src/main/java/com/genersoft/iot/vmp/gb28181/event/record/RecordEndEventListener.java

@@ -1,18 +1,18 @@
package com.genersoft.iot.vmp.gb28181.event.record;

import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
import com.genersoft.iot.vmp.gb28181.bean.RecordItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.*;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;

/**
 * @description: 录像查询结束时间
 * @description: 录像查询结束事件
 * @author: pan
 * @data: 2022-02-23
 */

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java

@@ -2,7 +2,6 @@


import com.alibaba.fastjson2.JSONObject;

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

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

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

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

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

@@ -12,45 +11,31 @@
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;

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

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

import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;

import com.genersoft.iot.vmp.gb28181.utils.SipUtils;

import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;

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

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

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

import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;

import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;

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

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

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.GitUtil;

import gov.nist.javax.sip.SIPConstants;

import gov.nist.javax.sip.SipProviderImpl;

import gov.nist.javax.sip.SipStackImpl;

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

import gov.nist.javax.sip.message.SIPRequest;

import gov.nist.javax.sip.message.SIPResponse;

import gov.nist.javax.sip.stack.SIPClientTransaction;

import gov.nist.javax.sip.stack.SIPClientTransactionImpl;

import gov.nist.javax.sip.stack.SIPDialog;

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.context.annotation.DependsOn;

import org.springframework.context.annotation.Lazy;

import org.springframework.stereotype.Component;

import org.springframework.util.ObjectUtils;



import javax.sip.*;

import javax.sip.address.Address;

import javax.sip.address.SipURI;

import javax.sip.header.*;

import javax.sip.message.Message;

import javax.sip.InvalidArgumentException;

import javax.sip.ResponseEvent;

import javax.sip.SipException;

import javax.sip.SipFactory;

import javax.sip.header.CallIdHeader;

import javax.sip.message.Request;

import javax.sip.message.Response;

import java.lang.reflect.Field;

import java.text.ParseException;

import java.util.HashSet;



/**

 * @description:设备能力接口，用于定义设备的控制、查询能力

@@ -183,7 +168,7 @@
    public void ptzCmd(Device device, String channelId, int leftRight, int upDown, int inOut, int moveSpeed,

                       int zoomSpeed) throws InvalidArgumentException, SipException, ParseException {

        String cmdStr = SipUtils.cmdString(leftRight, upDown, inOut, moveSpeed, zoomSpeed);

        StringBuffer ptzXml = new StringBuffer(200);

        StringBuilder ptzXml = new StringBuilder(200);

        String charset = device.getCharset();

        ptzXml.append("<?xml version=\"1.0\" encoding=\"" + charset + "\"?>\r\n");

        ptzXml.append("<Control>\r\n");


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
@@ -15,15 +14,12 @@
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import gov.nist.javax.sip.SipProviderImpl;
import gov.nist.javax.sip.message.MessageFactoryImpl;
import gov.nist.javax.sip.message.SIPRequest;
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.context.annotation.DependsOn;
import org.springframework.context.annotation.Lazy;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
@@ -638,7 +634,7 @@
        MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
        if (mediaServerItem != null) {
            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
            zlmrtpServerFactory.closeRTPServer(mediaServerItem, sendRtpItem.getStreamId());
            zlmrtpServerFactory.closeRtpServer(mediaServerItem, sendRtpItem.getStreamId());
        }
        SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(platform, sendRtpItem);
        if (byeRequest == null) {

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java

@@ -10,8 +10,8 @@
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg;
@@ -33,7 +33,8 @@
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import java.text.ParseException;
import java.util.*;
import java.util.HashMap;
import java.util.Map;

/**
 * SIP命令类型： ACK请求
@@ -61,6 +62,9 @@

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Autowired
    private ZlmHttpHookSubscribe hookSubscribe;

    @Autowired
    private IMediaServerService mediaServerService;
@@ -130,8 +134,18 @@
                startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
            });
        }else {
            JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
            // 如果是非严格模式，需要关闭端口占用
            JSONObject startSendRtpStreamResult = null;
            if (sendRtpItem.getLocalPort() != 0) {
                if (zlmrtpServerFactory.releasePort(mediaInfo, sendRtpItem.getSsrc())) {
                    startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                }
            }else {
                startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            }
            if (startSendRtpStreamResult != null) {
                startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader);
            }
        }
    }
    private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform,

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -45,6 +45,7 @@
import javax.sip.message.Response;
import java.text.ParseException;
import java.time.Instant;
import java.util.Random;
import java.util.Vector;

/**
@@ -157,11 +158,6 @@
                StreamProxyItem proxyByAppAndStream =null;
                // 不是通道可能是直播流
                if (channel != null && gbStream == null) {
//                    if (channel.getStatus() == 0) {
//                        logger.info("通道离线，返回400");
//                        responseAck(request, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline");
//                        return;
//                    }
                    // 通道存在，发100，TRYING
                    try {
                        responseAck(request, Response.TRYING);
@@ -385,7 +381,12 @@
                        } else {
                            content.append("t=0 0\r\n");
                        }
                        content.append("m=video " + sendRtpItem.getLocalPort() + " RTP/AVP 96\r\n");
                        int localPort = sendRtpItem.getLocalPort();
                        if (localPort == 0) {
                            // 非严格模式端口不统一, 增加兼容性，修改为一个不为0的端口
                            localPort = new Random().nextInt(65535) + 1;
                        }
                        content.append("m=video " + localPort + " RTP/AVP 96\r\n");
                        content.append("a=sendonly\r\n");
                        content.append("a=rtpmap:96 PS/90000\r\n");
                        content.append("y=" + sendRtpItem.getSsrc() + "\r\n");
@@ -476,9 +477,11 @@

                            // 写入redis， 超时时回复
                            redisCatchStorage.updateSendRTPSever(sendRtpItem);
                            MediaServerItem finalMediaServerItem = mediaServerItem;
                            playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg) -> {
                                logger.info("[上级点播]超时, 用户：{}， 通道：{}", username, channelId);
                                redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                                zlmrtpServerFactory.releasePort(finalMediaServerItem, sendRtpItem.getSsrc());
                            }, null);
                        } else {
                            sendRtpItem.setStreamId(playTransaction.getStream());

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

@@ -626,6 +626,32 @@
        return ret;

    }



    /**

     * rtpServer收流超时

     */

    @ResponseBody

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

    public JSONObject onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam param){

        System.out.println(param);

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



        JSONObject ret = new JSONObject();

        ret.put("code", 0);

        ret.put("msg", "success");



        taskExecutor.execute(()->{

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

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

                }

            }

        });



        return ret;

    }



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

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

        if (ObjectUtils.isEmpty(params)) {


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

@@ -1,15 +1,15 @@
package com.genersoft.iot.vmp.media.zlm;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.*;

@@ -23,6 +23,9 @@

    @Autowired
    private UserSetting userSetting;

    @Autowired
    private ZlmHttpHookSubscribe hookSubscribe;

    private int[] portRangeArray = new int[2];

@@ -141,7 +144,7 @@
        return result;
    }

    public boolean closeRTPServer(MediaServerItem serverItem, String streamId) {
    public boolean closeRtpServer(MediaServerItem serverItem, String streamId) {
        boolean result = false;
        if (serverItem !=null){
            Map<String, Object> param = new HashMap<>();
@@ -161,32 +164,6 @@
        return result;
    }

//    private int getPortFromportRange(MediaServerItem mediaServerItem) {
//        int currentPort = mediaServerItem.getCurrentPort();
//        if (currentPort == 0) {
//            String[] portRangeStrArray = mediaServerItem.getSendRtpPortRange().split(",");
//            if (portRangeStrArray.length != 2) {
//                portRangeArray[0] = 30000;
//                portRangeArray[1] = 30500;
//            }else {
//                portRangeArray[0] = Integer.parseInt(portRangeStrArray[0]);
//                portRangeArray[1] = Integer.parseInt(portRangeStrArray[1]);
//            }
//        }
//
//        if (currentPort == 0 || currentPort++ > portRangeArray[1]) {
//            currentPort = portRangeArray[0];
//            mediaServerItem.setCurrentPort(currentPort);
//            return portRangeArray[0];
//        } else {
//            if (currentPort % 2 == 1) {
//                currentPort++;
//            }
//            currentPort++;
//            mediaServerItem.setCurrentPort(currentPort);
//            return currentPort;
//        }
//    }

    /**
     * 创建一个国标推流
@@ -200,21 +177,15 @@
     */
    public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String deviceId, String channelId, boolean tcp){

        // 使用RTPServer 功能找一个可用的端口
        String sendRtpPortRange = serverItem.getSendRtpPortRange();
        if (ObjectUtils.isEmpty(sendRtpPortRange)) {
            return null;
        }
        String[] portRangeStrArray = serverItem.getSendRtpPortRange().split(",");
        int localPort = -1;
        if (portRangeStrArray.length != 2) {
            localPort = getFreePort(serverItem, 30000, 30500, null);
        }else {
            localPort = getFreePort(serverItem, Integer.parseInt(portRangeStrArray[0]),  Integer.parseInt(portRangeStrArray[1]), null);
        }
        if (localPort == -1) {
            logger.error("没有可用的端口");
            return null;
        // 默认为随机端口
        int localPort = 0;
        if (userSetting.getGbSendStreamStrict()) {
            if (userSetting.getGbSendStreamStrict()) {
                localPort = keepPort(serverItem, ssrc);
                if (localPort == 0) {
                    return null;
                }
            }
        }
        SendRtpItem sendRtpItem = new SendRtpItem();
        sendRtpItem.setIp(ip);
@@ -242,21 +213,13 @@
     * @return SendRtpItem
     */
    public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId, String app, String stream, String channelId, boolean tcp){
        // 使用RTPServer 功能找一个可用的端口
        String sendRtpPortRange = serverItem.getSendRtpPortRange();
        if (ObjectUtils.isEmpty(sendRtpPortRange)) {
            return null;
        }
        String[] portRangeStrArray = serverItem.getSendRtpPortRange().split(",");
        int localPort = -1;
        if (portRangeStrArray.length != 2) {
            localPort = getFreePort(serverItem, 30000, 30500, null);
        }else {
            localPort = getFreePort(serverItem, Integer.parseInt(portRangeStrArray[0]),  Integer.parseInt(portRangeStrArray[1]), null);
        }
        if (localPort == -1) {
            logger.error("没有可用的端口");
            return null;
        // 默认为随机端口
        int localPort = 0;
        if (userSetting.getGbSendStreamStrict()) {
            localPort = keepPort(serverItem, ssrc);
            if (localPort == 0) {
                return null;
            }
        }
        SendRtpItem sendRtpItem = new SendRtpItem();
        sendRtpItem.setIp(ip);
@@ -271,6 +234,42 @@
        sendRtpItem.setServerId(userSetting.getServerId());
        sendRtpItem.setMediaServerId(serverItem.getId());
        return sendRtpItem;
    }

    /**
     * 保持端口，直到需要需要发流时再释放
     */
    public int keepPort(MediaServerItem serverItem, String ssrc) {
        int localPort = 0;
        Map<String, Object> param = new HashMap<>(3);
        param.put("port", 0);
        param.put("enable_tcp", 1);
        param.put("stream_id", ssrc);
        JSONObject jsonObject = zlmresTfulUtils.openRtpServer(serverItem, param);
        if (jsonObject.getInteger("code") == 0) {
            localPort = jsonObject.getInteger("port");
            HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId());
            // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
            hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
                    (MediaServerItem mediaServerItem, JSONObject response)->{
                        logger.info("[上级点播] {}->监听端口到期继续保持监听", ssrc);
                        keepPort(serverItem, ssrc);
                    });
        }
        logger.info("[上级点播] {}->监听端口: {}", ssrc, localPort);
        return localPort;
    }

    /**
     * 释放保持的端口
     */
    public boolean releasePort(MediaServerItem serverItem, String ssrc) {
        logger.info("[上级点播] {}->释放监听端口，等待推流", ssrc);
        boolean closeRTPServerResult = closeRtpServer(serverItem, ssrc);
        HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId());
        // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
        hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
        return closeRTPServerResult;
    }

    /**
@@ -333,7 +332,7 @@
            result= true;
            logger.info("[停止RTP推流] 成功");
        } else {
            logger.error("[停止RTP推流] 失败: {}, 参数：{}->\r\n{}",jsonObject.getString("msg"),jsonObject.toJSONString(param));
            logger.error("[停止RTP推流] 失败: {}, 参数：{}->\r\n{}",jsonObject.getString("msg"), JSON.toJSON(param), jsonObject);
        }
        return result;
    }

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

@@ -18,7 +18,11 @@
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@Component
@Order(value=1)
@@ -73,8 +77,6 @@
            }
        });



        // 获取zlm信息
        logger.info("[zlm] 等待默认zlm中...");

@@ -87,7 +89,7 @@
        }
        for (MediaServerItem mediaServerItem : all) {
            if (startGetMedia == null) {
                startGetMedia = new HashMap<>();
                startGetMedia = new ConcurrentHashMap<>();
            }
            startGetMedia.put(mediaServerItem.getId(), true);
            connectZlmServer(mediaServerItem);
@@ -95,7 +97,7 @@
        }
        String taskKey = "zlm-connect-timeout";
        dynamicTask.startDelay(taskKey, ()->{
            if (startGetMedia != null) {
            if (startGetMedia != null && startGetMedia.size() > 0) {
                Set<String> allZlmId = startGetMedia.keySet();
                for (String id : allZlmId) {
                    logger.error("[ {} ]]主动连接失败，不再尝试连接", id);

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeFactory.java

@@ -24,6 +24,17 @@
        return hookSubscribe;
    }

    public static HookSubscribeForRtpServerTimeout on_rtp_server_timeout(String stream, String ssrc, String mediaServerId) {
        HookSubscribeForRtpServerTimeout hookSubscribe = new HookSubscribeForRtpServerTimeout();
        JSONObject subscribeKey = new com.alibaba.fastjson2.JSONObject();
        subscribeKey.put("stream_id", stream);
        subscribeKey.put("ssrc", ssrc);
        subscribeKey.put("mediaServerId", mediaServerId);
        hookSubscribe.setContent(subscribeKey);

        return hookSubscribe;
    }

    public static HookSubscribeForServerStarted on_server_started() {
        HookSubscribeForServerStarted hookSubscribe = new HookSubscribeForServerStarted();
        hookSubscribe.setContent(new JSONObject());

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeForRtpServerTimeout.java

New file
@@ -0,0 +1,44 @@
package com.genersoft.iot.vmp.media.zlm.dto;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.annotation.JSONField;

import java.time.Instant;

/**
 * hook订阅-收流超时
 * @author lin
 */
public class HookSubscribeForRtpServerTimeout implements IHookSubscribe{

    private HookType hookType = HookType.on_rtp_server_timeout;

    private JSONObject content;

    @JSONField(format="yyyy-MM-dd HH:mm:ss")
    private Instant expires;

    @Override
    public HookType getHookType() {
        return hookType;
    }

    @Override
    public JSONObject getContent() {
        return content;
    }

    public void setContent(JSONObject content) {
        this.content = content;
    }

    @Override
    public Instant getExpires() {
        return expires;
    }

    @Override
    public void setExpires(Instant expires) {
        this.expires = expires;
    }
}

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeForStreamChange.java

@@ -15,6 +15,7 @@

    private JSONObject content;

    @JSONField(format="yyyy-MM-dd HH:mm:ss")
    private Instant expires;

    @Override

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookType.java

@@ -19,5 +19,7 @@
    on_stream_none_reader,
    on_stream_not_found,
    on_server_started,

    on_rtp_server_timeout,
    on_server_keepalive
}

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java

@@ -65,9 +65,6 @@
    @Schema(description = "多端口RTP收流端口范围")
    private String rtpPortRange;

    @Schema(description = "RTP发流端口范围")
    private String sendRtpPortRange;

    @Schema(description = "assist服务端口")
    private int recordAssistPort;

@@ -118,7 +115,6 @@
        hookAliveInterval = zlmServerConfig.getHookAliveInterval();
        rtpEnable = false; // 默认使用单端口;直到用户自己设置开启多端口
        rtpPortRange = zlmServerConfig.getPortRange().replace("_",","); // 默认使用30000,30500作为级联时发送流的端口号
        sendRtpPortRange = "30000,30500"; // 默认使用30000,30500作为级联时发送流的端口号
        recordAssistPort = 0; // 默认关闭

    }
@@ -321,14 +317,6 @@

    public void setLastKeepaliveTime(String lastKeepaliveTime) {
        this.lastKeepaliveTime = lastKeepaliveTime;
    }

    public String getSendRtpPortRange() {
        return sendRtpPortRange;
    }

    public void setSendRtpPortRange(String sendRtpPortRange) {
        this.sendRtpPortRange = sendRtpPortRange;
    }

    public Float getHookAliveInterval() {

 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/OnRtpServerTimeoutHookParam.java

New file
@@ -0,0 +1,53 @@
package com.genersoft.iot.vmp.media.zlm.dto.hook;

/**
 * zlm hook事件中的on_rtp_server_timeout事件的参数
 * @author lin
 */
public class OnRtpServerTimeoutHookParam extends HookParam{
    private int local_port;
    private String stream_id;
    private int tcpMode;
    private boolean re_use_port;
    private String ssrc;

    public int getLocal_port() {
        return local_port;
    }

    public void setLocal_port(int local_port) {
        this.local_port = local_port;
    }

    public String getStream_id() {
        return stream_id;
    }

    public void setStream_id(String stream_id) {
        this.stream_id = stream_id;
    }

    public int getTcpMode() {
        return tcpMode;
    }

    public void setTcpMode(int tcpMode) {
        this.tcpMode = tcpMode;
    }

    public boolean isRe_use_port() {
        return re_use_port;
    }

    public void setRe_use_port(boolean re_use_port) {
        this.re_use_port = re_use_port;
    }

    public String getSsrc() {
        return ssrc;
    }

    public void setSsrc(String ssrc) {
        this.ssrc = ssrc;
    }
}

 src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java

@@ -4,13 +4,12 @@
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -24,12 +23,10 @@
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.support.incrementer.AbstractIdentityColumnMaxValueIncrementer;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
@@ -171,7 +168,7 @@
        redisCatchStorage.updateDevice(device);
        deviceMapper.update(device);
        //进行通道离线
        deviceChannelMapper.offlineByDeviceId(deviceId);
//        deviceChannelMapper.offlineByDeviceId(deviceId);
        // 离线释放所有ssrc
        List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(deviceId, null, null, null);
        if (ssrcTransactions != null && ssrcTransactions.size() > 0) {

 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java

@@ -168,7 +168,7 @@
        if (mediaServerItem == null) {
            return;
        }
        zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId);
        zlmrtpServerFactory.closeRtpServer(mediaServerItem, streamId);
        releaseSsrc(mediaServerItem.getId(), streamId);
    }

@@ -535,6 +535,7 @@
        param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
        param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
        param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex));
        param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrex));
        if (mediaServerItem.getRecordAssistPort() > 0) {
            param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
        }else {
@@ -545,8 +546,7 @@
        // 置0关闭此特性(推流断开会导致立即断开播放器)
        // 此参数不应大于播放器超时时间
        // 优化此消息以更快的收到流注销事件
        param.put("general.continue_push_ms", "3000" );
        param.put("general.publishToHls", "0" );
        param.put("protocol.continue_push_ms", "3000" );
        // 最多等待未初始化的Track时间，单位毫秒，超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流，
        // 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
//        param.put("general.wait_track_ready_ms", "3000" );

 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java

@@ -283,7 +283,7 @@
        try {
            cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
                logger.info("收到订阅消息： " + response.toJSONString());
                System.out.println("停止超时任务： " + timeOutTaskKey);
                dynamicTask.stop(timeOutTaskKey);

                // hook响应
                onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid);

 src/main/java/com/genersoft/iot/vmp/storager/dao/MediaServerMapper.java

@@ -28,7 +28,6 @@
            "secret, " +
            "rtpEnable, " +
            "rtpPortRange, " +
            "sendRtpPortRange, " +
            "recordAssistPort, " +
            "defaultServer, " +
            "createTime, " +
@@ -52,7 +51,6 @@
            "'${secret}', " +
            "${rtpEnable}, " +
            "'${rtpPortRange}', " +
            "'${sendRtpPortRange}', " +
            "${recordAssistPort}, " +
            "${defaultServer}, " +
            "'${createTime}', " +
@@ -77,7 +75,6 @@
            "<if test=\"autoConfig != null\">, autoConfig=${autoConfig}</if>" +
            "<if test=\"rtpEnable != null\">, rtpEnable=${rtpEnable}</if>" +
            "<if test=\"rtpPortRange != null\">, rtpPortRange='${rtpPortRange}'</if>" +
            "<if test=\"sendRtpPortRange != null\">, sendRtpPortRange='${sendRtpPortRange}'</if>" +
            "<if test=\"secret != null\">, secret='${secret}'</if>" +
            "<if test=\"recordAssistPort != null\">, recordAssistPort=${recordAssistPort}</if>" +
            "<if test=\"hookAliveInterval != null\">, hookAliveInterval=${hookAliveInterval}</if>" +
@@ -101,7 +98,6 @@
            "<if test=\"autoConfig != null\">, autoConfig=${autoConfig}</if>" +
            "<if test=\"rtpEnable != null\">, rtpEnable=${rtpEnable}</if>" +
            "<if test=\"rtpPortRange != null\">, rtpPortRange='${rtpPortRange}'</if>" +
            "<if test=\"sendRtpPortRange != null\">, sendRtpPortRange='${sendRtpPortRange}'</if>" +
            "<if test=\"secret != null\">, secret='${secret}'</if>" +
            "<if test=\"recordAssistPort != null\">, recordAssistPort=${recordAssistPort}</if>" +
            "<if test=\"hookAliveInterval != null\">, hookAliveInterval=${hookAliveInterval}</if>" +

 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java

@@ -62,7 +62,9 @@
        if (callId != null) {
            // 权限校验
            StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
            if (streamAuthorityInfo.getCallId().equals(callId)) {
            if (streamAuthorityInfo != null
                    && streamAuthorityInfo.getCallId() != null
                    && streamAuthorityInfo.getCallId().equals(callId)) {
                authority = true;
            }else {
                throw new ControllerException(ErrorCode.ERROR400);

 src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java

@@ -135,14 +135,8 @@
        MediaServerItem mediaServerItemInDatabase = mediaServerService.getOne(mediaServerItem.getId());

        if (mediaServerItemInDatabase != null) {
            if (ObjectUtils.isEmpty(mediaServerItemInDatabase.getSendRtpPortRange()) && ObjectUtils.isEmpty(mediaServerItem.getSendRtpPortRange())) {
                mediaServerItem.setSendRtpPortRange("30000,30500");
            }
            mediaServerService.update(mediaServerItem);
        } else {
            if (ObjectUtils.isEmpty(mediaServerItem.getSendRtpPortRange())) {
                mediaServerItem.setSendRtpPortRange("30000,30500");
            }
            mediaServerService.add(mediaServerItem);
        }
    }

 src/main/resources/all-application.yml

@@ -192,6 +192,9 @@
    stream-on-demand: true
    # 推流鉴权， 默认开启
    push-authority: true
    # 国标级联发流严格模式，严格模式会使用与sdp信息中一致的端口发流，端口共享media.rtp.port-range，这会损失一些性能，
    # 非严格模式使用随机端口发流，性能更好， 默认关闭
    gb-send-stream-strict: false

# 关闭在线文档（生产环境建议关闭）
springdoc:

 web_src/src/components/dialog/MediaServerEdit.vue

@@ -89,11 +89,6 @@
                -
                <el-input v-model="rtpPortRange2" placeholder="终止" @change="portRangeChange" clearable style="width: 100px" prop="rtpPortRange2" :disabled="mediaServerForm.defaultServer"></el-input>
              </el-form-item>
              <el-form-item label="推流端口" prop="sendRtpPortRange1">
                <el-input v-model="sendRtpPortRange1" placeholder="起始" @change="portRangeChange" clearable style="width: 100px" prop="sendRtpPortRange1" :disabled="mediaServerForm.defaultServer"></el-input>
                -
                <el-input v-model="sendRtpPortRange2" placeholder="终止" @change="portRangeChange" clearable style="width: 100px" prop="sendRtpPortRange2" :disabled="mediaServerForm.defaultServer"></el-input>
              </el-form-item>
              <el-form-item label="录像管理服务端口" prop="recordAssistPort">
                <el-input v-model.number="mediaServerForm.recordAssistPort" :disabled="mediaServerForm.defaultServer">
<!--                  <el-button v-if="mediaServerForm.recordAssistPort > 0" slot="append" type="primary" @click="checkRecordServer">测试</el-button>-->
@@ -177,15 +172,12 @@
        rtmpSSlPort: "",
        rtpEnable: false,
        rtpPortRange: "",
        sendRtpPortRange: "",
        rtpProxyPort: "",
        rtspPort: "",
        rtspSSLPort: "",
      },
      rtpPortRange1:30000,
      rtpPortRange2:30500,
      sendRtpPortRange1:30000,
      sendRtpPortRange2:30500,

      rules: {
        ip:  [{ required: true, validator: isValidIp, message: '请输入有效的IP地址', trigger: 'blur' }],
@@ -196,8 +188,6 @@
        rtmpSSlPort:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        rtpPortRange1:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        rtpPortRange2:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        sendRtpPortRange1:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        sendRtpPortRange2:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        rtpProxyPort:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        rtspPort:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
        rtspSSLPort:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
@@ -229,9 +219,6 @@
            this.rtpPortRange2 =  rtpPortRange[1]
          }
        }
        let sendRtpPortRange = this.mediaServerForm.sendRtpPortRange.split(",");
        this.sendRtpPortRange1 = sendRtpPortRange[0]
        this.sendRtpPortRange2 = sendRtpPortRange[1]
      }
    },
    checkServer: function() {
@@ -251,8 +238,6 @@
          that.mediaServerForm = data.data;
          that.mediaServerForm.httpPort = httpPort;
          that.mediaServerForm.autoConfig = true;
          that.sendRtpPortRange1 = 30000
          that.sendRtpPortRange2 = 30500
          that.rtpPortRange1 = 30000
          that.rtpPortRange2 = 30500
          that.serverCheck = 1;
@@ -336,13 +321,10 @@
        rtmpSSlPort: "",
        rtpEnable: false,
        rtpPortRange: "",
        sendRtpPortRange: "",
        rtpProxyPort: "",
        rtspPort: "",
        rtspSSLPort: "",
      };
      this.sendRtpPortRange1 = 30000;
      this.sendRtpPortRange2 = 30500;
      this.rtpPortRange1 = 30500;
      this.rtpPortRange2 = 30500;
      this.listChangeCallback = null
@@ -367,9 +349,7 @@
      }
    },
    portRangeChange: function() {
      this.mediaServerForm.sendRtpPortRange = this.sendRtpPortRange1 + "," + this.sendRtpPortRange2
      this.mediaServerForm.rtpPortRange = this.rtpPortRange1 + "," + this.rtpPortRange2
      console.log(this.mediaServerForm.sendRtpPortRange)
      console.log(this.mediaServerForm.rtpPortRange)
    }
  },