合并主线的发流端口管理逻辑

648540858

2023-07-10 04e7f48fde1b1a653d413eb41186ec7354f5ae31

合并主线的发流端口管理逻辑

 sql/2.6.8²¹¶¡¸üÐÂ.sql

New file
@@ -0,0 +1,2 @@
alter table media_server
    add sendRtpPortRange varchar(50) not null;

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

@@ -75,6 +75,9 @@
    @Value("${media.rtp.port-range}")
    private String rtpPortRange;

    @Value("${media.rtp.send-port-range}")
    private String rtpSendPortRange;

    @Value("${media.record-assist-port:0}")
    private Integer recordAssistPort = 0;

@@ -206,6 +209,7 @@
        mediaServerItem.setSecret(secret);
        mediaServerItem.setRtpEnable(rtpEnable);
        mediaServerItem.setRtpPortRange(rtpPortRange);
        mediaServerItem.setSendRtpPortRange(rtpSendPortRange);
        mediaServerItem.setRecordAssistPort(recordAssistPort);
        mediaServerItem.setHookAliveInterval(30.00f);

@@ -222,4 +226,11 @@
        return false;
    }

    public String getRtpSendPortRange() {
        return rtpSendPortRange;
    }

    public void setRtpSendPortRange(String rtpSendPortRange) {
        this.rtpSendPortRange = rtpSendPortRange;
    }
}

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

@@ -140,15 +140,7 @@
                startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
            });
        }else {
            // å¦‚果是非严格模式，需要关闭端口占用
            JSONObject startSendRtpStreamResult = null;
            if (sendRtpItem.getLocalPort() != 0) {
                if (zlmrtpServerFactory.releasePort(mediaInfo, sendRtpItem.getSsrc())) {
                    startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                }
            }else {
                startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            }
            JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            if (startSendRtpStreamResult != null) {
                startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader);
            }

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

@@ -349,9 +349,7 @@
                    }
                    logger.info("[上级Invite] {}, å¹³å°ï¼š{}， é€šé“：{}, æ”¶æµåœ°å€ï¼š{}:{}，收流方式：{}, ssrc：{}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                            device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback -> {
                                return redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
                            });
                            device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());

                    if (tcpActive != null) {
                        sendRtpItem.setTcpActive(tcpActive);
@@ -553,9 +551,7 @@
            if (streamReady) {
                // è‡ªå¹³å°å†…容
                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback ->{
                            return  redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
                        });
                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());

                if (sendRtpItem == null) {
                    logger.warn("服务器端口资源不足");
@@ -594,9 +590,7 @@
            if (streamReady) {
                // è‡ªå¹³å°å†…容
                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback ->{
                            return  redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
                        });
                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());

                if (sendRtpItem == null) {
                    logger.warn("服务器端口资源不足");
@@ -713,9 +707,7 @@
                dynamicTask.stop(callIdHeader.getCallId());
                if (serverId.equals(userSetting.getServerId())) {
                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
                            app, stream, channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback -> {
                                return redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
                            });
                            app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());

                    if (sendRtpItem == null) {
                        logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足");

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

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

import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.media.zlm.dto.MediaSendRtpPortInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class SendRtpPortManager {

    private final static Logger logger = LoggerFactory.getLogger(SendRtpPortManager.class);

    @Autowired
    private UserSetting userSetting;

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    private final String KEY = "VM_MEDIA_SEND_RTP_PORT_RANGE_";


    public void initServerPort(String mediaServerId, int startPort, int endPort){
        String key = KEY + userSetting.getServerId() + "_" +  mediaServerId;
        MediaSendRtpPortInfo mediaSendRtpPortInfo = new MediaSendRtpPortInfo(startPort, endPort, mediaServerId);
        redisTemplate.opsForValue().set(key, mediaSendRtpPortInfo);
    }

    public int getNextPort(String mediaServerId) {
        String key = KEY + userSetting.getServerId() + "_" +  mediaServerId;
        MediaSendRtpPortInfo mediaSendRtpPortInfo = (MediaSendRtpPortInfo)redisTemplate.opsForValue().get(key);
        if (mediaSendRtpPortInfo == null) {
            logger.warn("[发送端口管理] èŽ·å–{}的发送端口时未找到端口信息", mediaSendRtpPortInfo);
            return 0;
        }
        int port;
        if (mediaSendRtpPortInfo.getCurrent() %2 != 0) {
            port = mediaSendRtpPortInfo.getCurrent() + 1;
        }else {
            port = mediaSendRtpPortInfo.getCurrent() + 2;
        }
        if (port > mediaSendRtpPortInfo.getEnd()) {
            if (mediaSendRtpPortInfo.getStart() %2 != 0) {
                port = mediaSendRtpPortInfo.getStart() + 1;
            }else {
                port = mediaSendRtpPortInfo.getStart();
            }
        }
        mediaSendRtpPortInfo.setCurrent(port);
        redisTemplate.opsForValue().set(key, mediaSendRtpPortInfo);
        return port;
    }
}

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

@@ -23,7 +23,6 @@
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;

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

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

import com.genersoft.iot.vmp.utils.redis.RedisUtil;

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

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

@@ -223,9 +222,6 @@




        HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();

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

            result.setEnable_audio(true);

        }



        taskExecutor.execute(() -> {

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

@@ -259,20 +255,6 @@
            }

        }



        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "*";

        // å°†ä¿¡æ¯å†™å…¥redis中，以备后用

        List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey);

        if (scan.size()>0) {

            for (Object o : scan) {

                String key = (String) o;

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

                if (otherRtpSendInfo != null && otherRtpSendInfo.getStream().equalsIgnoreCase(param.getStream())) {

                    result.setEnable_audio(true);

                    result.setEnable_mp4(true);

                }

            }

        }



        if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {

            logger.info("推流时发现尚未设置录像路径，从assist服务中读取");

            JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);

@@ -291,6 +273,18 @@
                }

            }

        }

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

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

            System.out.println(receiveKey);

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

            System.out.println("otherRtpSendInfo != null ====>" + (otherRtpSendInfo != null));

            if (otherRtpSendInfo != null) {

                System.out.println("otherRtpSendInfo != null");

                result.setEnable_audio(true);

                result.setEnable_mp4(true);

            }

        }

        logger.info("[ZLM HOOK]推流鉴权 å“åº”：{}->{}->>>>{}", param.getMediaServerId(), param, result);

        return result;

    }




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

@@ -1,20 +1,18 @@
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.common.CommonCallback;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.HashMap;
import java.util.Map;

@Component
public class ZLMRTPServerFactory {
@@ -30,68 +28,8 @@
    @Autowired
    private ZlmHttpHookSubscribe hookSubscribe;

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

    public int getFreePort(MediaServerItem mediaServerItem, int startPort, int endPort, List<Integer> usedFreelist) {
        if (endPort <= startPort) {
            return -1;
        }
        if (usedFreelist == null) {
            usedFreelist = new ArrayList<>();
        }
        JSONObject listRtpServerJsonResult = zlmresTfulUtils.listRtpServer(mediaServerItem);
        if (listRtpServerJsonResult != null) {
            JSONArray data = listRtpServerJsonResult.getJSONArray("data");
            if (data != null) {
                for (int i = 0; i < data.size(); i++) {
                    JSONObject dataItem = data.getJSONObject(i);
                    usedFreelist.add(dataItem.getInteger("port"));
                }
            }
        }

        Map<String, Object> param = new HashMap<>();
        int result = -1;
        // è®¾ç½®æŽ¨æµç«¯å£
        if (startPort%2 == 1) {
            startPort ++;
        }
        boolean checkPort = false;
        for (int i = startPort; i < endPort  + 1; i+=2) {
            if (!usedFreelist.contains(i)){
                checkPort = true;
                startPort = i;
                break;
            }
        }
        if (!checkPort) {
            logger.warn("未找到节点{}上范围[{}-{}]的空闲端口", mediaServerItem.getId(), startPort, endPort);
            return -1;
        }
        param.put("port", startPort);
        String stream = UUID.randomUUID().toString();
        param.put("enable_tcp", 1);
        param.put("stream_id", stream);
//        param.put("port", 0);
        JSONObject openRtpServerResultJson = zlmresTfulUtils.openRtpServer(mediaServerItem, param);

        if (openRtpServerResultJson != null) {
            if (openRtpServerResultJson.getInteger("code") == 0) {
                result= openRtpServerResultJson.getInteger("port");
                Map<String, Object> closeRtpServerParam = new HashMap<>();
                closeRtpServerParam.put("stream_id", stream);
                zlmresTfulUtils.closeRtpServer(mediaServerItem, closeRtpServerParam);
            }else {
                usedFreelist.add(startPort);
                startPort +=2;
                result = getFreePort(mediaServerItem, startPort, endPort,usedFreelist);
            }
        }else {
            //  æ£€æŸ¥ZLM状态
            logger.error("创建RTP Server å¤±è´¥ {}: è¯·æ£€æŸ¥ZLM服务", param.get("port"));
        }
        return result;
    }
    @Autowired
    private SendRtpPortManager sendRtpPortManager;

    /**
     * å¼€å¯rtpServer
@@ -220,13 +158,13 @@
     * @return SendRtpItem
     */
    public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId,
                                         String deviceId, String channelId, boolean tcp, boolean rtcp, KeepPortCallback callback){
                                         String deviceId, String channelId, boolean tcp, boolean rtcp){

        // é»˜è®¤ä¸ºéšæœºç«¯å£
        int localPort = 0;
        if (userSetting.getGbSendStreamStrict()) {
            if (userSetting.getGbSendStreamStrict()) {
                localPort = keepPort(serverItem, ssrc, localPort, callback);
                localPort = sendRtpPortManager.getNextPort(serverItem.getId());
                if (localPort == 0) {
                    return null;
                }
@@ -259,11 +197,11 @@
     * @return SendRtpItem
     */
    public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId,
                                         String app, String stream, String channelId, boolean tcp, boolean rtcp, KeepPortCallback callback){
                                         String app, String stream, String channelId, boolean tcp, boolean rtcp){
        // é»˜è®¤ä¸ºéšæœºç«¯å£
        int localPort = 0;
        if (userSetting.getGbSendStreamStrict()) {
            localPort = keepPort(serverItem, ssrc, localPort, callback);
            localPort = sendRtpPortManager.getNextPort(serverItem.getId());
            if (localPort == 0) {
                return null;
            }
@@ -282,58 +220,6 @@
        sendRtpItem.setMediaServerId(serverItem.getId());
        sendRtpItem.setRtcp(rtcp);
        return sendRtpItem;
    }

    public interface KeepPortCallback{
        Boolean keep(String ssrc);
    }

    /**
     * ä¿æŒç«¯å£ï¼Œç›´åˆ°éœ€è¦éœ€è¦å‘流时再释放
     */
    public int keepPort(MediaServerItem serverItem, String ssrc, int localPort, KeepPortCallback keepPortCallback) {
        Map<String, Object> param = new HashMap<>(3);
        param.put("port", localPort);
        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也会从这里进入系统
            Integer finalLocalPort = localPort;
            hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
                    (MediaServerItem mediaServerItem, JSONObject response)->{
                        System.out.println("监听端口到期继续保持监听");
                        System.out.println(response);
                        if (ssrc.equals(response.getString("stream_id"))) {
                            if (keepPortCallback.keep(ssrc)) {
                                logger.info("[上级点播] {}->监听端口到期继续保持监听", ssrc);
                                keepPort(serverItem, ssrc, finalLocalPort, keepPortCallback);
                            }else {
                                logger.info("[上级点播] {}->发送取消，无需继续监听", ssrc);
                                releasePort(serverItem, ssrc);
                            }
                        }

                    });
            logger.info("[上级点播] {}->监听端口: {}", ssrc, localPort);
        }else {
            logger.info("[上级点播] ç›‘听端口失败: {}", ssrc);
        }
        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;
    }

    /**

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

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

public class MediaSendRtpPortInfo {

    private int start;
    private int end;
    private String mediaServerId;

    private int current;


    public MediaSendRtpPortInfo(int start, int end, String mediaServerId) {
        this.start = start;
        this.current = start;
        this.end = end;
        this.mediaServerId = mediaServerId;
    }

    public int getStart() {
        return start;
    }

    public void setStart(int start) {
        this.start = start;
    }

    public int getEnd() {
        return end;
    }

    public void setEnd(int end) {
        this.end = end;
    }

    public String getMediaServerId() {
        return mediaServerId;
    }

    public void setMediaServerId(String mediaServerId) {
        this.mediaServerId = mediaServerId;
    }

    public int getCurrent() {
        return current;
    }

    public void setCurrent(int current) {
        this.current = current;
    }
}

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

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

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

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

@@ -297,4 +300,12 @@
    public void setHookAliveInterval(Float hookAliveInterval) {
        this.hookAliveInterval = hookAliveInterval;
    }

    public String getSendRtpPortRange() {
        return sendRtpPortRange;
    }

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

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

@@ -50,4 +50,14 @@
    public void setMp4_save_path(String mp4_save_path) {
        this.mp4_save_path = mp4_save_path;
    }

    @Override
    public String toString() {
        return "HookResultForOnPublish{" +
                "enable_audio=" + enable_audio +
                ", enable_mp4=" + enable_mp4 +
                ", mp4_max_second=" + mp4_max_second +
                ", mp4_save_path='" + mp4_save_path + '\'' +
                '}';
    }
}

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

@@ -11,10 +11,7 @@
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.*;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -70,6 +67,10 @@
    private UserSetting userSetting;

    @Autowired
    private SendRtpPortManager sendRtpPortManager;


    @Autowired
    private AssistRESTfulUtils assistRESTfulUtils;

    @Autowired
@@ -115,13 +116,40 @@
            if (ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) {
                ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null);
            }
            if (userSetting.getGbSendStreamStrict()) {
                int startPort = 50000;
                int endPort = 60000;
                String sendRtpPortRange = mediaServerItem.getSendRtpPortRange();
                if (sendRtpPortRange == null) {
                    logger.warn("[zlm] ] æœªé…ç½®å‘流端口范围，默认使用50000到60000");
                }else {
                    String[] sendRtpPortRangeArray = sendRtpPortRange.trim().split(",");
                    if (sendRtpPortRangeArray.length != 2) {
                        logger.warn("[zlm] ] å‘流端口范围错误，默认使用50000到60000");
                    }else {
                        try {
                            startPort = Integer.parseInt(sendRtpPortRangeArray[0]);
                            endPort = Integer.parseInt(sendRtpPortRangeArray[1]);
                            if (endPort <= startPort) {
                                logger.warn("[zlm] ] å‘流端口范围错误，结束端口应大于开始端口,使用默认端口");
                                startPort = 50000;
                                endPort = 60000;
                            }

                        }catch (NumberFormatException e) {
                            logger.warn("[zlm] ] å‘流端口范围错误，默认使用50000到60000");
                        }
                    }
                }
                logger.info("[[zlm] ] é…ç½®å‘流端口范围，{}-{}", startPort, endPort);
                sendRtpPortManager.initServerPort(mediaServerItem.getId(), startPort, endPort);
            }
            // æŸ¥è¯¢redis是否存在此mediaServer
            String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
            Boolean hasKey = redisTemplate.hasKey(key);
            if (hasKey != null && ! hasKey) {
                redisTemplate.opsForValue().set(key, mediaServerItem);
            }

        }
    }


 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java

@@ -317,9 +317,7 @@
        SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
                content.getPort(), content.getSsrc(), content.getPlatformId(),
                content.getApp(), content.getStream(), content.getChannelId(),
                content.getTcp(), content.getRtcp(), ssrcFromCallback -> {
                    return querySendRTPServer(content.getPlatformId(), content.getChannelId(), content.getStream(), null) != null;
                });
                content.getTcp(), content.getRtcp());

        WVPResult<ResponseSendItemMsg> result = new WVPResult<>();
        result.setCode(0);

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

@@ -28,6 +28,7 @@
            "secret, " +
            "rtpEnable, " +
            "rtpPortRange, " +
            "sendRtpPortRange, " +
            "recordAssistPort, " +
            "defaultServer, " +
            "createTime, " +
@@ -51,6 +52,7 @@
            "#{secret}, " +
            "#{rtpEnable}, " +
            "#{rtpPortRange}, " +
            "#{sendRtpPortRange}, " +
            "#{recordAssistPort}, " +
            "#{defaultServer}, " +
            "#{createTime}, " +
@@ -75,6 +77,7 @@
            "<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>" +
@@ -98,6 +101,7 @@
            "<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/bean/OtherRtpSendInfo.java

@@ -124,7 +124,7 @@
    @Override
    public String toString() {
        return "OtherRtpSendInfo{" +
                "ip='" + ip + '\'' +
                "  ip='" + ip + '\'' +
                ", port=" + port +
                ", receiveIp='" + receiveIp + '\'' +
                ", receivePort=" + receivePort +

 src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java

@@ -3,20 +3,16 @@
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
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.VersionInfo;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
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.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
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.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo;
import io.swagger.v3.oas.annotations.Operation;
@@ -27,7 +23,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*;
@@ -36,6 +31,7 @@
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@SuppressWarnings("rawtypes")
@Tag(name = "第三方服务对接")
@@ -56,19 +52,10 @@
    private IMediaServerService mediaServerService;

    @Autowired
    private VersionInfo versionInfo;

    @Autowired
    private SipConfig sipConfig;
    private SendRtpPortManager sendRtpPortManager;

    @Autowired
    private UserSetting userSetting;

    @Autowired
    private IDeviceService deviceService;

    @Autowired
    private IDeviceChannelService channelService;

    @Autowired
    private DynamicTask dynamicTask;
@@ -76,14 +63,6 @@

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;


    @Value("${server.port}")
    private int serverPort;


    @Autowired
    private IRedisCatchStorage redisCatchStorage;


    @GetMapping(value = "/receive/open")
@@ -145,24 +124,15 @@
        otherRtpSendInfo.setReceivePort(localPort);
        otherRtpSendInfo.setCallId(callId);
        otherRtpSendInfo.setStream(stream);
        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + stream;
        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_"  + stream;
        // å°†ä¿¡æ¯å†™å…¥redis中，以备后用
        redisTemplate.opsForValue().set(receiveKey, otherRtpSendInfo);
        if (isSend != null && isSend) {
            String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId;
            String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
            // é¢„创建发流信息
            int port = zlmServerFactory.keepPort(mediaServerItem, callId, 0, ssrc1 -> {
                return redisTemplate.opsForValue().get(key) != null;
            });

            int port = sendRtpPortManager.getNextPort(mediaServerItem.getId());
            // å°†ä¿¡æ¯å†™å…¥redis中，以备后用
            redisTemplate.opsForValue().set(key, otherRtpSendInfo);
            // è®¾ç½®è¶…时任务，超时未使用，则自动移除，并关闭端口保持, é»˜è®¤äº”分钟
            dynamicTask.startDelay(key, ()->{
                logger.info("[第三方服务对接->开启收流和获取发流信息] ç«¯å£ä¿æŒè¶…æ—¶ callId->{}", callId);
                redisTemplate.delete(key);
                zlmServerFactory.releasePort(mediaServerItem, callId);
            }, 15000);
            redisTemplate.opsForValue().set(key, otherRtpSendInfo, 300, TimeUnit.SECONDS);
            otherRtpSendInfo.setIp(mediaServerItem.getSdpIp());
            otherRtpSendInfo.setPort(port);
            logger.info("[第三方服务对接->开启收流和获取发流信息] ç»“果，callId->{}， {}", callId, otherRtpSendInfo);
@@ -178,7 +148,7 @@
        logger.info("[第三方服务对接->关闭收流] stream->{}", stream);
        MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
        zlmServerFactory.closeRtpServer(mediaServerItem,stream);
        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + stream;
        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_"  + stream;
        // å°†ä¿¡æ¯å†™å…¥redis中，以备后用
        redisTemplate.delete(receiveKey);
    }
@@ -203,11 +173,9 @@
            streamType = 1;
        }
        MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId;
        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
        OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
        if (sendInfo != null) {
            zlmServerFactory.releasePort(mediaServerItem, callId);
        }else {
        if (sendInfo == null) {
            sendInfo = new OtherRtpSendInfo();
        }
        sendInfo.setPushApp(app);
@@ -229,7 +197,6 @@
        param.put("only_audio", onlyAudio ? "1" : "0");
        param.put("pt", pt);

        dynamicTask.stop(key);
        Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream);
        if (streamReady) {
            logger.info("[第三方服务对接->发送流] æµå­˜åœ¨ï¼Œå¼€å§‹å‘流，callId->{}", callId);
@@ -279,7 +246,7 @@
    @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true)
    public void closeSendRTP(String callId) {
        logger.info("[第三方服务对接->关闭发送流] callId->{}", callId);
        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId;
        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
        OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
        if (sendInfo == null){
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流");

 web_src/src/components/dialog/MediaServerEdit.vue

@@ -89,6 +89,11 @@
                -
                <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 v-if="mediaServerForm.sendRtpEnable" label="发流端口" >
                <el-input v-model="sendRtpPortRange1" placeholder="起始" @change="portRangeChange" clearable style="width: 100px" prop="rtpPortRange1" :disabled="mediaServerForm.defaultServer"></el-input>
                -
                <el-input v-model="sendRtpPortRange2" placeholder="终止" @change="portRangeChange" clearable style="width: 100px" prop="rtpPortRange2" :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>-->
@@ -172,12 +177,16 @@
        rtmpSSlPort: "",
        rtpEnable: false,
        rtpPortRange: "",
        sendRtpPortRange: "",
        rtpProxyPort: "",
        rtspPort: "",
        rtspSSLPort: "",
      },
      rtpPortRange1:30000,
      rtpPortRange2:30500,

      sendRtpPortRange1:50000,
      sendRtpPortRange2:60000,

      rules: {
        ip:  [{ required: true, validator: isValidIp, message: '请输入有效的IP地址', trigger: 'blur' }],
@@ -214,9 +223,14 @@
        this.currentStep = 3;
        if (param.rtpPortRange) {
          let rtpPortRange = this.mediaServerForm.rtpPortRange.split(",");
          let sendRtpPortRange = this.mediaServerForm.sendRtpPortRange.split(",");
          if (rtpPortRange.length > 0) {
            this.rtpPortRange1 =  rtpPortRange[0]
            this.rtpPortRange2 =  rtpPortRange[1]
          }
          if (sendRtpPortRange.length > 0) {
            this.sendRtpPortRange1 =  sendRtpPortRange[0]
            this.sendRtpPortRange2 =  sendRtpPortRange[1]
          }
        }
      }
@@ -240,6 +254,8 @@
          that.mediaServerForm.autoConfig = true;
          that.rtpPortRange1 = 30000
          that.rtpPortRange2 = 30500
          that.sendRtpPortRange1 = 50000
          that.sendRtpPortRange2 = 60000
          that.serverCheck = 1;
        }else {
          that.serverCheck = -1;
@@ -321,12 +337,15 @@
        rtmpSSlPort: "",
        rtpEnable: false,
        rtpPortRange: "",
        sendRtpPortRange: "",
        rtpProxyPort: "",
        rtspPort: "",
        rtspSSLPort: "",
      };
      this.rtpPortRange1 = 30500;
      this.rtpPortRange2 = 30500;
      this.sendRtpPortRange1 = 50000;
      this.sendRtpPortRange2 = 60000;
      this.listChangeCallback = null
      this.currentStep = 1;
    },
@@ -351,7 +370,7 @@
    portRangeChange: function() {
      if (this.mediaServerForm.rtpEnable) {
        this.mediaServerForm.rtpPortRange = this.rtpPortRange1 + "," + this.rtpPortRange2
        console.log(this.mediaServerForm.rtpPortRange)
        this.mediaServerForm.sendRtpPortRange = this.sendRtpPortRange1 + "," + this.sendRtpPortRange2
      }
    }
  },