From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 29 五月 2024 15:02:51 +0800
Subject: [PATCH] 调整上级观看消息的发送
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java | 72 +++++++++++++++++------------------
1 files changed, 35 insertions(+), 37 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
index 0713ab3..d3f69d6 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
@@ -9,20 +9,16 @@
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
-import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
-import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
-import com.genersoft.iot.vmp.media.event.MediaNotFoundEvent;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.event.hook.Hook;
+import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
+import com.genersoft.iot.vmp.media.event.hook.HookType;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
-import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
-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.media.zlm.dto.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
-import com.genersoft.iot.vmp.media.zlm.dto.hook.HookResult;
-import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.IGbStreamService;
-import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -46,7 +42,6 @@
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
-import org.springframework.web.context.request.async.DeferredResult;
import java.util.HashMap;
import java.util.List;
@@ -66,12 +61,6 @@
@Autowired
private IVideoManagerStorage videoManagerStorager;
-
- @Autowired
- private IMediaService mediaService;
-
- @Autowired
- private ZLMServerFactory zlmServerFactory;
@Autowired
private StreamProxyMapper streamProxyMapper;
@@ -98,7 +87,7 @@
private IMediaServerService mediaServerService;
@Autowired
- private ZlmHttpHookSubscribe hookSubscribe;
+ private HookSubscribe hookSubscribe;
@Autowired
private DynamicTask dynamicTask;
@@ -205,9 +194,9 @@
callback.run(ErrorCode.ERROR100.getCode(), "淇濆瓨澶辫触", null);
return;
}
- HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaServer.getId());
- hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+ Hook hook = Hook.getInstance(HookType.on_media_arrival, param.getApp(), param.getStream(), mediaServer.getId());
+ hookSubscribe.addSubscribe(hook, (hookData) -> {
+ StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
mediaServer, param.getApp(), param.getStream(), null, null);
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
});
@@ -215,7 +204,7 @@
String talkKey = UUID.randomUUID().toString();
String delayTalkKey = UUID.randomUUID().toString();
dynamicTask.startDelay(delayTalkKey, ()->{
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaServer.getId(), false);
+ StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaServer.getId(), false);
if (streamInfo != null) {
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
}else {
@@ -225,9 +214,9 @@
}, 7000);
WVPResult<String> result = addStreamProxyToZlm(param);
if (result != null && result.getCode() == 0) {
- hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
+ hookSubscribe.removeSubscribe(hook);
dynamicTask.stop(talkKey);
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+ StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
mediaServer, param.getApp(), param.getStream(), null, null);
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
}else {
@@ -246,7 +235,7 @@
}
}
else{
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+ StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
mediaServer, param.getApp(), param.getStream(), null, null);
callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
}
@@ -355,11 +344,14 @@
if (mediaServer == null) {
return null;
}
- if (zlmServerFactory.isStreamReady(mediaServer, param.getApp(), param.getStream())) {
+ if (mediaServerService.isStreamReady(mediaServer, param.getApp(), param.getStream())) {
mediaServerService.closeStreams(mediaServer, param.getApp(), param.getStream());
}
String msgResult;
if ("ffmpeg".equalsIgnoreCase(param.getType())){
+ if (param.getTimeoutMs() == 0) {
+ param.setTimeoutMs(15);
+ }
result = mediaServerService.addFFmpegSource(mediaServer, param.getSrcUrl().trim(), param.getDstUrl(),
param.getTimeoutMs(), param.isEnableAudio(), param.isEnableMp4(),
param.getFfmpegCmdKey());
@@ -367,8 +359,6 @@
result = mediaServerService.addStreamProxy(mediaServer, param.getApp(), param.getStream(), param.getUrl().trim(),
param.isEnableAudio(), param.isEnableMp4(), param.getRtpType());
}
- System.out.println("addStreamProxyToZlm====");
- System.out.println(result);
if (result != null && result.getCode() == 0) {
String key = result.getData();
if (key == null) {
@@ -387,6 +377,13 @@
return null;
}
MediaServer mediaServer = mediaServerService.getOne(param.getMediaServerId());
+ if (mediaServer == null) {
+ return null;
+ }
+ List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaServer, param.getApp(), param.getStream(), null);
+ if (mediaList == null || mediaList.isEmpty()) {
+ return true;
+ }
Boolean result = false;
if ("ffmpeg".equalsIgnoreCase(param.getType())){
result = mediaServerService.delFFmpegSource(mediaServer, param.getStreamKey());
@@ -412,6 +409,7 @@
gbStreamMapper.del(app, stream);
videoManagerStorager.deleteStreamProxy(app, stream);
redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream);
+ redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PUSH", app, stream);
Boolean result = removeStreamProxyFromZlm(streamProxyItem);
if (result != null && result) {
logger.info("[绉婚櫎浠ｇ悊]锛� 浠ｇ悊锛� {}/{}, 浠巣lm绉婚櫎鎴愬姛", app, stream);
@@ -510,18 +508,18 @@
String type = "PULL";
// 鍙戦�乺edis娑堟伅
- List<OnStreamChangedHookParam> onStreamChangedHookParams = redisCatchStorage.getStreams(mediaServerId, type);
- if (onStreamChangedHookParams.size() > 0) {
- for (OnStreamChangedHookParam onStreamChangedHookParam : onStreamChangedHookParams) {
+ List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+ if (mediaInfoList.size() > 0) {
+ for (MediaInfo mediaInfo : mediaInfoList) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetting.getServerId());
- jsonObject.put("app", onStreamChangedHookParam.getApp());
- jsonObject.put("stream", onStreamChangedHookParam.getStream());
+ jsonObject.put("app", mediaInfo.getApp());
+ jsonObject.put("stream", mediaInfo.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerId, type, onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream());
+ redisCatchStorage.removeStream(mediaServerId, type, mediaInfo.getApp(), mediaInfo.getStream());
}
}
}
@@ -539,8 +537,8 @@
private void syncPullStream(String mediaServerId){
MediaServer mediaServer = mediaServerService.getOne(mediaServerId);
if (mediaServer != null) {
- List<OnStreamChangedHookParam> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
- if (!allPullStream.isEmpty()) {
+ List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, "PULL");
+ if (!mediaInfoList.isEmpty()) {
List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaServer, null, null, null);
Map<String, StreamInfo> stringStreamInfoMap = new HashMap<>();
if (mediaList != null && !mediaList.isEmpty()) {
--
Gitblit v1.8.0