From 1de344674afd6bb35b51b165bbad76dbe6299b7e Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 28 三月 2024 18:08:33 +0800 Subject: [PATCH] 使用冒号分隔redis的key --- src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java | 174 ++++++++++++++++++++++++++++++++++++++++++++------------- 1 files changed, 134 insertions(+), 40 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 d63781b..59c5ace --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java @@ -2,13 +2,16 @@ import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; +import com.baomidou.dynamic.datasource.annotation.DS; 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; import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; +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; @@ -27,25 +30,31 @@ import com.genersoft.iot.vmp.storager.dao.StreamProxyMapper; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; -import com.genersoft.iot.vmp.vmanager.bean.ResourceBaceInfo; +import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo; import com.github.pagehelper.PageInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionStatus; +import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; /** * 瑙嗛浠g悊涓氬姟 */ @Service +@DS("master") public class StreamProxyServiceImpl implements IStreamProxyService { private final static Logger logger = LoggerFactory.getLogger(StreamProxyServiceImpl.class); @@ -58,6 +67,9 @@ @Autowired private ZLMRESTfulUtils zlmresTfulUtils; + + @Autowired + private ZLMServerFactory zlmServerFactory; @Autowired private StreamProxyMapper streamProxyMapper; @@ -78,12 +90,6 @@ private PlatformGbStreamMapper platformGbStreamMapper; @Autowired - private EventPublisher eventPublisher; - - @Autowired - private ParentPlatformMapper parentPlatformMapper; - - @Autowired private IGbStreamService gbStreamService; @Autowired @@ -91,6 +97,9 @@ @Autowired private ZlmHttpHookSubscribe hookSubscribe; + + @Autowired + private DynamicTask dynamicTask; @Autowired DataSourceTransactionManager dataSourceTransactionManager; @@ -119,7 +128,13 @@ } JSONArray dataArray = jsonObject.getJSONArray("data"); JSONObject mediaServerConfig = dataArray.getJSONObject(0); - String ffmpegCmd = mediaServerConfig.getString(param.getFfmpeg_cmd_key()); + if (ObjectUtils.isEmpty(param.getFfmpegCmdKey())) { + param.setFfmpegCmdKey("ffmpeg.cmd"); + } + String ffmpegCmd = mediaServerConfig.getString(param.getFfmpegCmdKey()); + if (ffmpegCmd == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(), "ffmpeg鎷夋祦浠g悊鏃犳硶鑾峰彇ffmpeg cmd"); + } String schema = getSchemaFromFFmpegCmd(ffmpegCmd); if (schema == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "ffmpeg鎷夋祦浠g悊鏃犳硶浠巉fmpeg cmd涓幏鍙栧埌杈撳嚭鏍煎紡"); @@ -130,9 +145,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 { @@ -143,10 +155,10 @@ dstUrl = String.format("%s://%s:%s/%s/%s", schemaForUri, "127.0.0.1", port, param.getApp(), param.getStream()); }else { - dstUrl = String.format("rtmp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtmpPort(), param.getApp(), + dstUrl = String.format("rtsp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtspPort(), param.getApp(), param.getStream()); } - param.setDst_url(dstUrl); + param.setDstUrl(dstUrl); logger.info("[鎷夋祦浠g悊] 杈撳嚭鍦板潃涓猴細{}", dstUrl); param.setMediaServerId(mediaInfo.getId()); boolean saveResult; @@ -160,37 +172,50 @@ 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(); + 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); }else { param.setEnable(false); // 鐩存帴绉婚櫎 - if (param.isEnable_remove_none_reader()) { + if (param.isEnableRemoveNoneReader()) { del(param.getApp(), param.getStream()); }else { updateStreamProxy(param); } 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); } } @@ -297,14 +322,32 @@ if (mediaServerItem == null) { return null; } - if ("default".equals(param.getType())){ - result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl(), - param.isEnable_audio(), param.isEnable_mp4(), param.getRtp_type()); - }else if ("ffmpeg".equals(param.getType())) { - result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrc_url(), param.getDst_url(), - param.getTimeout_ms() + "", param.isEnable_audio(), param.isEnable_mp4(), - param.getFfmpeg_cmd_key()); - System.out.println(result); + if (zlmServerFactory.isStreamReady(mediaServerItem, param.getApp(), param.getStream())) { + zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream()); + } + if ("ffmpeg".equalsIgnoreCase(param.getType())){ + result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrcUrl().trim(), param.getDstUrl(), + param.getTimeoutMs() + "", param.isEnableAudio(), param.isEnableMp4(), + param.getFfmpegCmdKey()); + }else { + result = zlmresTfulUtils.addStreamProxy(mediaServerItem, 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.getInteger("code") == 0) { + JSONObject data = result.getJSONObject("data"); + if (data == null) { + logger.warn("[鑾峰彇鎷夋祦浠g悊鐨勭粨鏋滄暟鎹瓺ata] 澶辫触锛� {}", result ); + return result; + } + String key = data.getString("key"); + if (key == null) { + logger.warn("[鑾峰彇鎷夋祦浠g悊鐨勭粨鏋滄暟鎹瓺ata涓殑KEY] 澶辫触锛� {}", result ); + return result; + } + param.setStreamKey(key); + streamProxyMapper.update(param); } return result; } @@ -315,7 +358,12 @@ return null; } MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId()); - JSONObject result = zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream()); + JSONObject result = null; + if ("ffmpeg".equalsIgnoreCase(param.getType())){ + result = zlmresTfulUtils.delFFmpegSource(mediaServerItem, param.getStreamKey()); + }else { + result = zlmresTfulUtils.delStreamProxy(mediaServerItem, param.getStreamKey()); + } return result; } @@ -329,18 +377,19 @@ StreamProxyItem streamProxyItem = videoManagerStorager.queryStreamProxy(app, stream); if (streamProxyItem != null) { gbStreamService.sendCatalogMsg(streamProxyItem, CatalogEvent.DEL); + + // 濡傛灉鍏宠仈浜嗗浗鏍囬偅涔堢Щ闄ゅ叧鑱� + platformGbStreamMapper.delByAppAndStream(app, stream); + gbStreamMapper.del(app, stream); videoManagerStorager.deleteStreamProxy(app, stream); + redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream); JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem); if (jsonObject != null && jsonObject.getInteger("code") == 0) { - // 濡傛灉鍏宠仈浜嗗浗鏍囬偅涔堢Щ闄ゅ叧鑱� - gbStreamMapper.del(app, stream); - platformGbStreamMapper.delByAppAndStream(app, stream); - // TODO 濡傛灉鍏宠仈鐨勬帹娴侊紝 閭d箞鐘舵�佽缃负绂荤嚎 + logger.info("[绉婚櫎浠g悊]锛� 浠g悊锛� {}/{}, 浠巣lm绉婚櫎鎴愬姛", app, stream); + }else { + logger.info("[绉婚櫎浠g悊]锛� 浠g悊锛� {}/{}, 浠巣lm绉婚櫎澶辫触", app, stream); } - redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream); } - - } @Override @@ -358,8 +407,10 @@ updateStreamProxy(streamProxy); }else { logger.info("鍚敤浠g悊澶辫触锛� {}/{}->{}({})", app, stream, jsonObject.getString("msg"), - streamProxy.getSrc_url() == null? streamProxy.getUrl():streamProxy.getSrc_url()); + streamProxy.getSrcUrl() == null? streamProxy.getUrl():streamProxy.getSrcUrl()); } + } else if (streamProxy != null && streamProxy.isEnable()) { + return true ; } return result; } @@ -404,14 +455,14 @@ @Override public void zlmServerOnline(String mediaServerId) { // 绉婚櫎寮�鍚簡鏃犱汉瑙傜湅鑷姩绉婚櫎鐨勬祦 - List<StreamProxyItem> streamProxyItemList = streamProxyMapper.selecAutoRemoveItemByMediaServerId(mediaServerId); + List<StreamProxyItem> streamProxyItemList = streamProxyMapper.selectAutoRemoveItemByMediaServerId(mediaServerId); if (streamProxyItemList.size() > 0) { gbStreamMapper.batchDel(streamProxyItemList); } streamProxyMapper.deleteAutoRemoveItemByMediaServerId(mediaServerId); // 绉婚櫎鎷夋祦浠g悊鐢熸垚鐨勬祦淇℃伅 -// syncPullStream(mediaServerId); + syncPullStream(mediaServerId); // 鎭㈠娴佷唬鐞�, 鍙煡鎵捐繖涓繖涓祦濯掍綋 List<StreamProxyItem> streamProxyListForEnable = storager.getStreamProxyListForEnableInMediaServer( @@ -432,7 +483,7 @@ @Override public void zlmServerOffline(String mediaServerId) { // 绉婚櫎寮�鍚簡鏃犱汉瑙傜湅鑷姩绉婚櫎鐨勬祦 - List<StreamProxyItem> streamProxyItemList = streamProxyMapper.selecAutoRemoveItemByMediaServerId(mediaServerId); + List<StreamProxyItem> streamProxyItemList = streamProxyMapper.selectAutoRemoveItemByMediaServerId(mediaServerId); if (streamProxyItemList.size() > 0) { gbStreamMapper.batchDel(streamProxyItemList); } @@ -510,7 +561,50 @@ } @Override - public ResourceBaceInfo getOverview() { - return streamProxyMapper.getOverview(); + public ResourceBaseInfo getOverview() { + + int total = streamProxyMapper.getAllCount(); + int online = streamProxyMapper.getOnline(); + + return new ResourceBaseInfo(total, online); + } + + + @Scheduled(cron = "* 0/10 * * * ?") + public void asyncCheckStreamProxyStatus() { + + List<MediaServerItem> all = mediaServerService.getAllOnline(); + + if (CollectionUtils.isEmpty(all)){ + return; + } + + Map<String, MediaServerItem> serverItemMap = all.stream().collect(Collectors.toMap(MediaServerItem::getId, Function.identity(), (m1, m2) -> m1)); + + List<StreamProxyItem> list = videoManagerStorager.getStreamProxyListForEnable(true); + + if (CollectionUtils.isEmpty(list)){ + return; + } + + for (StreamProxyItem streamProxyItem : list) { + + MediaServerItem mediaServerItem = serverItemMap.get(streamProxyItem.getMediaServerId()); + + // TODO 鏀寔鍏朵粬 schema + JSONObject mediaInfo = zlmresTfulUtils.isMediaOnline(mediaServerItem, streamProxyItem.getApp(), streamProxyItem.getStream(), "rtsp"); + + if (mediaInfo == null){ + streamProxyItem.setStatus(false); + } else { + if (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")) { + streamProxyItem.setStatus(true); + } else { + streamProxyItem.setStatus(false); + } + } + + updateStreamProxy(streamProxyItem); + } } } -- Gitblit v1.8.0