Merge branch 'wvp-28181-2.0' into main-dev

648540858

2023-10-11 7de73ebd2bc07a51f0f9db031d6f616bdcfe549c

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

old mode 100644
new mode 100755
@@ -24,23 +24,30 @@
import com.genersoft.iot.vmp.utils.JsonUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.RecordFile;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
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.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;

import java.io.File;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 媒体服务器节点管理
@@ -87,7 +94,7 @@


    @Autowired
    private ZLMServerFactory ZLMServerFactory;
    private ZLMServerFactory zlmServerFactory;

    @Autowired
    private EventPublisher publisher;
@@ -104,6 +111,11 @@
    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    @Qualifier("taskExecutor")
    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;





@@ -118,36 +130,8 @@
                continue;
            }
            // 更新
            if (ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) {
            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();
@@ -179,11 +163,16 @@
        }

        if (streamId == null) {
            streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
            streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase();
        }
        int ssrcCheckParam = 0;
        if (ssrcCheck && tcpMode > 1) {
            // 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验
            logger.warn("[openRTPServer] TCP被动/TCP主动收流时，默认关闭ssrc检验");
        }
        int rtpServerPort;
        if (mediaServerItem.isRtpEnable()) {
            rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port, onlyAuto, reUsePort, tcpMode);
            rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0) ? Long.parseLong(ssrc) : 0, port, onlyAuto, reUsePort, tcpMode);
        } else {
            rtpServerPort = mediaServerItem.getRtpProxyPort();
        }
@@ -201,7 +190,7 @@
        if (mediaServerItem == null) {
            return;
        }
        ZLMServerFactory.closeRtpServer(mediaServerItem, streamId);
        zlmServerFactory.closeRtpServer(mediaServerItem, streamId);
    }

    @Override
@@ -210,7 +199,7 @@
            callback.run(false);
            return;
        }
        ZLMServerFactory.closeRtpServer(mediaServerItem, streamId, callback);
        zlmServerFactory.closeRtpServer(mediaServerItem, streamId, callback);
    }

    @Override
@@ -221,7 +210,7 @@

    @Override
    public Boolean updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) {
        return ZLMServerFactory.updateRtpServerSSRC(mediaServerItem, streamId, ssrc);
        return zlmServerFactory.updateRtpServerSSRC(mediaServerItem, streamId, ssrc);
    }

    @Override
@@ -248,7 +237,7 @@
        mediaServerMapper.update(mediaSerItem);
        MediaServerItem mediaServerItemInRedis = getOne(mediaSerItem.getId());
        MediaServerItem mediaServerItemInDataBase = mediaServerMapper.queryOne(mediaSerItem.getId());
        if (mediaServerItemInRedis == null || ssrcFactory.hasMediaServerSSRC(mediaSerItem.getId())) {
        if (mediaServerItemInRedis == null || !ssrcFactory.hasMediaServerSSRC(mediaSerItem.getId())) {
            ssrcFactory.initMediaServerSSRC(mediaServerItemInDataBase.getId(),null);
        }
        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItemInDataBase.getId();
@@ -431,7 +420,7 @@
        }
        mediaServerMapper.update(serverItem);
        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + zlmServerConfig.getGeneralMediaServerId();
        if (ssrcFactory.hasMediaServerSSRC(serverItem.getId())) {
        if (!ssrcFactory.hasMediaServerSSRC(serverItem.getId())) {
            ssrcFactory.initMediaServerSSRC(zlmServerConfig.getGeneralMediaServerId(), null);
        }
        redisTemplate.opsForValue().set(key, serverItem);
@@ -781,4 +770,89 @@
        return result;
    }

    @Override
    public List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
        Assert.notNull(app, "app不存在");
        Assert.notNull(stream, "stream不存在");
        Assert.notNull(startTime, "startTime不存在");
        Assert.notNull(endTime, "endTime不存在");
        Assert.notEmpty(mediaServerItems, "流媒体列表为空");

        CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];
        for (int i = 0; i < mediaServerItems.size(); i++) {
            completableFutures[i] = getRecordFilesForOne(app, stream, startTime, endTime, mediaServerItems.get(i));
        }
        List<RecordFile> result = new ArrayList<>();
        for (int i = 0; i < completableFutures.length; i++) {
            try {
                List<RecordFile> list = (List<RecordFile>) completableFutures[i].get();
                if (!list.isEmpty()) {
                    for (int g = 0; g < list.size(); g++) {
                        list.get(g).setMediaServerId(mediaServerItems.get(i).getId());
                    }
                    result.addAll(list);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        Comparator<RecordFile> comparator = Comparator.comparing(RecordFile::getFileName);
        result.sort(comparator);
        return result;
    }

    @Override
    public List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems) {
        Assert.notNull(app, "app不存在");
        Assert.notNull(stream, "stream不存在");
        Assert.notEmpty(mediaServerItems, "流媒体列表为空");
        CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];

        for (int i = 0; i < mediaServerItems.size(); i++) {
            completableFutures[i] = getRecordDatesForOne(app, stream, year, month, mediaServerItems.get(i));
        }
        List<String> result = new ArrayList<>();
        CompletableFuture.allOf(completableFutures).join();
        for (CompletableFuture completableFuture : completableFutures) {
            try {
                List<String> list = (List<String>) completableFuture.get();
                result.addAll(list);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        Collections.sort(result);
        return result;
    }

    @Async
    public CompletableFuture<List<String>> getRecordDatesForOne(String app, String stream, int year, int month, MediaServerItem mediaServerItem) {
        JSONObject fileListJson = assistRESTfulUtils.getDateList(mediaServerItem, app, stream, year, month);
        if (fileListJson != null && !fileListJson.isEmpty()) {
            if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
                JSONArray data = fileListJson.getJSONArray("data");
                return CompletableFuture.completedFuture(data.toJavaList(String.class));
            }
        }
        return CompletableFuture.completedFuture(new ArrayList<>());
    }

    @Async
    public CompletableFuture<List<RecordFile>> getRecordFilesForOne(String app, String stream, String startTime, String endTime, MediaServerItem mediaServerItem) {
        JSONObject fileListJson = assistRESTfulUtils.getFileList(mediaServerItem, 1, 100000000, app, stream, startTime, endTime);
        if (fileListJson != null && !fileListJson.isEmpty()) {
            if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
                JSONObject data = fileListJson.getJSONObject("data");
                JSONArray list = data.getJSONArray("list");
                if (list != null) {
                    return CompletableFuture.completedFuture(list.toJavaList(RecordFile.class));
                }
            }
        }
        return CompletableFuture.completedFuture(new ArrayList<>());
    }
}