From 483d5e04a71b6632419699df1077ac5a5457ad38 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期一, 25 九月 2023 18:51:02 +0800
Subject: [PATCH] 优化拉流代理
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java | 45 ++++++++++++++++++++++++++++++++++-----------
1 files changed, 34 insertions(+), 11 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
old mode 100644
new mode 100755
index 9ac6ab9..f36cff9
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
@@ -4,6 +4,7 @@
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.GeneralCallback;
import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
@@ -41,6 +42,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
/**
* 瑙嗛浠ｇ悊涓氬姟
@@ -93,6 +95,9 @@
private ZlmHttpHookSubscribe hookSubscribe;
@Autowired
+ private DynamicTask dynamicTask;
+
+ @Autowired
DataSourceTransactionManager dataSourceTransactionManager;
@Autowired
@@ -130,9 +135,6 @@
port = mediaInfo.getRtspPort();
schemaForUri = schema;
}else if (schema.equalsIgnoreCase("flv")) {
- port = mediaInfo.getHttpPort();
- schemaForUri = "http";
- }else if (schema.equalsIgnoreCase("rtmp")) {
port = mediaInfo.getRtmpPort();
schemaForUri = schema;
}else {
@@ -160,18 +162,34 @@
callback.run(ErrorCode.ERROR100.getCode(), "淇濆瓨澶辫触", null);
return;
}
-
HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaInfo.getId());
hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
mediaInfo, param.getApp(), param.getStream(), null, null);
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
});
-
if (param.isEnable()) {
+ String talkKey = UUID.randomUUID().toString();
+// dynamicTask.startCron(talkKey, ()->{
+// StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaInfo.getId(), false);
+// if (streamInfo != null) {
+// callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
+// }
+// }, 3000);
+ String delayTalkKey = UUID.randomUUID().toString();
+ dynamicTask.startDelay(delayTalkKey, ()->{
+ StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaInfo.getId(), false);
+ if (streamInfo != null) {
+ callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
+ }else {
+ dynamicTask.stop(talkKey);
+ callback.run(ErrorCode.ERROR100.getCode(), "瓒呮椂", null);
+ }
+ }, 7000);
JSONObject jsonObject = addStreamProxyToZlm(param);
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
+ dynamicTask.stop(talkKey);
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
mediaInfo, param.getApp(), param.getStream(), null, null);
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
@@ -185,12 +203,15 @@
}
if (jsonObject == null){
callback.run(ErrorCode.ERROR100.getCode(), "璁板綍宸蹭繚瀛橈紝鍚敤澶辫触", null);
- return;
}else {
callback.run(ErrorCode.ERROR100.getCode(), jsonObject.getString("msg"), null);
- return;
}
}
+ }
+ else{
+ StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+ mediaInfo, param.getApp(), param.getStream(), null, null);
+ callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
}
}
@@ -298,10 +319,10 @@
return null;
}
if ("default".equals(param.getType())){
- result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl(),
+ result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl().trim(),
param.isEnableAudio(), param.isEnableMp4(), param.getRtpType());
}else if ("ffmpeg".equals(param.getType())) {
- result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrcUrl(), param.getDstUrl(),
+ result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrcUrl().trim(), param.getDstUrl(),
param.getTimeoutMs() + "", param.isEnableAudio(), param.isEnableMp4(),
param.getFfmpegCmdKey());
}
@@ -328,14 +349,16 @@
StreamProxyItem streamProxyItem = videoManagerStorager.queryStreamProxy(app, stream);
if (streamProxyItem != null) {
gbStreamService.sendCatalogMsg(streamProxyItem, CatalogEvent.DEL);
- videoManagerStorager.deleteStreamProxy(app, stream);
+
JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
// 濡傛灉鍏宠仈浜嗗浗鏍囬偅涔堢Щ闄ゅ叧鑱�
+ int i = platformGbStreamMapper.delByAppAndStream(app, stream);
gbStreamMapper.del(app, stream);
- platformGbStreamMapper.delByAppAndStream(app, stream);
+ System.out.println();
// TODO 濡傛灉鍏宠仈鐨勬帹娴侊紝 閭ｄ箞鐘舵�佽缃负绂荤嚎
}
+ videoManagerStorager.deleteStreamProxy(app, stream);
redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream);
}
--
Gitblit v1.8.0