From fc77b3f819b3143387b90a4d631725e7c6513ecd Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 01 十二月 2023 15:49:18 +0800
Subject: [PATCH] 支持重新接入zlm的时候检查拉流代理数据是否异常，异常数据自动移除
---
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java | 137 ++++++++++++++++++++++++---------------------
1 files changed, 72 insertions(+), 65 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
old mode 100644
new mode 100755
index 856359c..7c4965e
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
@@ -3,6 +3,7 @@
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.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
@@ -10,12 +11,10 @@
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.IInviteStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
@@ -25,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;
/**
* 濯掍綋鏈嶅姟鍣ㄨ妭鐐圭鐞�
@@ -69,6 +75,9 @@
private UserSetting userSetting;
@Autowired
+ private SendRtpPortManager sendRtpPortManager;
+
+ @Autowired
private AssistRESTfulUtils assistRESTfulUtils;
@Autowired
@@ -85,7 +94,7 @@
@Autowired
- private ZLMRTPServerFactory zlmrtpServerFactory;
+ private ZLMServerFactory zlmServerFactory;
@Autowired
private EventPublisher publisher;
@@ -97,7 +106,15 @@
private IRedisCatchStorage redisCatchStorage;
@Autowired
+ private IInviteStreamService inviteStreamService;
+
+ @Autowired
private RedisTemplate<Object, Object> redisTemplate;
+
+ @Qualifier("taskExecutor")
+ @Autowired
+ private ThreadPoolTaskExecutor taskExecutor;
+
/**
@@ -111,7 +128,7 @@
continue;
}
// 鏇存柊
- if (ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) {
+ if (!ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) {
ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null);
}
// 鏌ヨredis鏄惁瀛樺湪姝ediaServer
@@ -120,17 +137,13 @@
if (hasKey != null && ! hasKey) {
redisTemplate.opsForValue().set(key, mediaServerItem);
}
-
}
}
- @Override
- public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck, boolean isPlayback) {
- return openRTPServer(mediaServerItem, streamId, null, ssrcCheck,isPlayback);
- }
@Override
- public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck, boolean isPlayback, Integer port) {
+ public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck,
+ boolean isPlayback, Integer port, Boolean reUsePort, Integer tcpMode) {
if (mediaServerItem == null || mediaServerItem.getId() == null) {
logger.info("[openRTPServer] 澶辫触, mediaServerItem == null || mediaServerItem.getId() == null");
return null;
@@ -148,11 +161,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锛屾殏鏃跺叧闂璼src鏍￠獙
+ logger.warn("[openRTPServer] TCP琚姩/TCP涓诲姩鏀舵祦鏃讹紝榛樿鍏抽棴ssrc妫�楠�");
}
int rtpServerPort;
if (mediaServerItem.isRtpEnable()) {
- rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port);
+ rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0) ? Long.parseLong(ssrc) : 0, port, reUsePort, tcpMode);
} else {
rtpServerPort = mediaServerItem.getRtpProxyPort();
}
@@ -160,22 +178,31 @@
}
@Override
- public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String ssrc, boolean ssrcCheck, boolean isPlayback) {
- return openRTPServer(mediaServerItem, streamId, ssrc, ssrcCheck, isPlayback, null);
- }
-
- @Override
public void closeRTPServer(MediaServerItem mediaServerItem, String streamId) {
if (mediaServerItem == null) {
return;
}
- zlmrtpServerFactory.closeRtpServer(mediaServerItem, streamId);
+ zlmServerFactory.closeRtpServer(mediaServerItem, streamId);
+ }
+
+ @Override
+ public void closeRTPServer(MediaServerItem mediaServerItem, String streamId, CommonCallback<Boolean> callback) {
+ if (mediaServerItem == null) {
+ callback.run(false);
+ return;
+ }
+ zlmServerFactory.closeRtpServer(mediaServerItem, streamId, callback);
}
@Override
public void closeRTPServer(String mediaServerId, String streamId) {
MediaServerItem mediaServerItem = this.getOne(mediaServerId);
closeRTPServer(mediaServerItem, streamId);
+ }
+
+ @Override
+ public Boolean updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) {
+ return zlmServerFactory.updateRtpServerSSRC(mediaServerItem, streamId, ssrc);
}
@Override
@@ -202,7 +229,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();
@@ -384,7 +411,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);
@@ -392,22 +419,11 @@
if (serverItem.isAutoConfig()) {
- // 鏌ョ湅assist鏈嶅姟鐨勫綍鍍忚矾寰勯厤缃�
- if (serverItem.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {
- JSONObject info = assistRESTfulUtils.getInfo(serverItem, null);
- if (info != null && info.getInteger("code") != null && info.getInteger("code") == 0 ) {
- JSONObject dataJson = info.getJSONObject("data");
- if (dataJson != null) {
- String recordPath = dataJson.getString("record");
- userSetting.setRecordPath(recordPath);
- }
- }
- }
setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));
}
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId();
dynamicTask.stop(zlmKeepaliveKey);
- dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (Math.getExponent(serverItem.getHookAliveInterval()) + 5) * 1000);
+ dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (serverItem.getHookAliveInterval().intValue() + 5) * 1000);
publisher.zlmOnlineEventPublish(serverItem.getId());
logger.info("[ZLM] 杩炴帴鎴愬姛 {} - {}:{} ",
@@ -537,7 +553,7 @@
logger.info("[ZLM] 姝ｅ湪璁剧疆 锛歿} -> {}:{}",
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
String protocol = sslEnabled ? "https" : "http";
- String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
+ String hookPrefix = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
@@ -546,25 +562,21 @@
}
param.put("hook.enable","1");
param.put("hook.on_flow_report","");
- param.put("hook.on_play",String.format("%s/on_play", hookPrex));
+ param.put("hook.on_play",String.format("%s/on_play", hookPrefix));
param.put("hook.on_http_access","");
- param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
+ param.put("hook.on_publish", String.format("%s/on_publish", hookPrefix));
param.put("hook.on_record_ts","");
param.put("hook.on_rtsp_auth","");
param.put("hook.on_rtsp_realm","");
- param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrex));
+ param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrefix));
param.put("hook.on_shell_login","");
- param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
- param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
- 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 {
- param.put("hook.on_record_mp4","");
- }
+ param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrefix));
+ param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrefix));
+ param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrefix));
+ param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrefix));
+ param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrefix));
+ param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrefix));
+ param.put("hook.on_record_mp4",String.format("%s/on_record_mp4", hookPrefix));
param.put("hook.timeoutSec","20");
// 鎺ㄦ祦鏂紑鍚庡彲浠ュ湪瓒呮椂鏃堕棿鍐呴噸鏂拌繛鎺ヤ笂缁х画鎺ㄦ祦锛岃繖鏍锋挱鏀惧櫒浼氭帴鐫�鎾斁銆�
// 缃�0鍏抽棴姝ょ壒鎬�(鎺ㄦ祦鏂紑浼氬鑷寸珛鍗虫柇寮�鎾斁鍣�)
@@ -573,15 +585,13 @@
param.put("protocol.continue_push_ms", "3000" );
// 鏈�澶氱瓑寰呮湭鍒濆鍖栫殑Track鏃堕棿锛屽崟浣嶆绉掞紝瓒呮椂涔嬪悗浼氬拷鐣ユ湭鍒濆鍖栫殑Track, 璁剧疆姝ら�夐」浼樺寲閭ｄ簺闊抽閿欒鐨勪笉瑙勮寖娴侊紝
// 绛墇lm鏀寔缁欐瘡涓猺tpServer璁剧疆鍏抽棴闊抽鐨勬椂鍊欏彲浠ヤ笉璁剧疆姝ら�夐」
-// param.put("general.wait_track_ready_ms", "3000" );
if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
}
- if (userSetting.getRecordPath() != null) {
- File recordPathFile = new File(userSetting.getRecordPath());
- File mp4SavePathFile = recordPathFile.getParentFile().getAbsoluteFile();
- param.put("protocol.mp4_save_path", mp4SavePathFile.getAbsoluteFile());
+ if (mediaServerItem.getRecordPath() != null) {
+ File recordPathFile = new File(mediaServerItem.getRecordPath());
+ param.put("protocol.mp4_save_path", recordPathFile.getParentFile().getPath());
param.put("record.appName", recordPathFile.getName());
}
@@ -678,7 +688,7 @@
// 缂撳瓨涓嶅瓨鍦紝浠庢暟鎹簱鏌ヨ锛屽鏋滄暟鎹簱涓嶅瓨鍦ㄥ垯鏄敊璇殑
mediaServerItem = getOneFromDatabase(mediaServerId);
if (mediaServerItem == null) {
- logger.warn("[鏇存柊ZLM 淇濇椿淇℃伅]澶辫触锛屾湭鎵惧埌娴佸獟浣撲俊鎭�");
+ logger.warn("[鏇存柊ZLM 淇濇椿淇℃伅] 娴佸獟浣搟}灏氭湭鍔犲叆浣跨敤,璇锋鏌ヨ妭鐐逛腑鏄惁鍚湁姝ゆ祦濯掍綋 ", mediaServerId);
return;
}
// zlm杩炴帴閲嶈瘯
@@ -714,22 +724,19 @@
}
@Override
- public boolean checkRtpServer(MediaServerItem mediaServerItem, String app, String stream) {
- JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, stream);
- if(rtpInfo.getInteger("code") == 0){
- return rtpInfo.getBoolean("exist");
- }
- return false;
- }
-
- @Override
public MediaServerLoad getLoad(MediaServerItem mediaServerItem) {
MediaServerLoad result = new MediaServerLoad();
result.setId(mediaServerItem.getId());
result.setPush(redisCatchStorage.getPushStreamCount(mediaServerItem.getId()));
result.setProxy(redisCatchStorage.getProxyStreamCount(mediaServerItem.getId()));
- result.setGbReceive(redisCatchStorage.getGbReceiveCount(mediaServerItem.getId()));
+
+ result.setGbReceive(inviteStreamService.getStreamInfoCount(mediaServerItem.getId()));
result.setGbSend(redisCatchStorage.getGbSendCount(mediaServerItem.getId()));
return result;
}
+
+ @Override
+ public List<MediaServerItem> getAllWithAssistPort() {
+ return mediaServerMapper.queryAllWithAssistPort();
+ }
}
--
Gitblit v1.8.0