From ad36354ef46a31f24b2583263f575d6736c0ad28 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 05 十二月 2023 11:29:05 +0800
Subject: [PATCH] 优化国标录像服务端，使用zlm新接口实现功能
---
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 267 +++++++++++++++++++++++++++++++---------------------
1 files changed, 158 insertions(+), 109 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
old mode 100644
new mode 100755
index 260b9a4..1e8dee7
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -1,5 +1,7 @@
package com.genersoft.iot.vmp.service.impl;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus;
@@ -12,7 +14,6 @@
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
-import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
@@ -22,24 +23,21 @@
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;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.*;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
-import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
-import com.genersoft.iot.vmp.service.bean.SSRCInfo;
+import com.genersoft.iot.vmp.service.bean.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import gov.nist.javax.sip.message.SIPResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
@@ -77,16 +75,13 @@
private IInviteStreamService inviteStreamService;
@Autowired
- private DeferredResultHolder resultHolder;
+ private ZlmHttpHookSubscribe subscribe;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private ZLMServerFactory zlmServerFactory;
-
- @Autowired
- private AssistRESTfulUtils assistRESTfulUtils;
@Autowired
private IMediaService mediaService;
@@ -107,7 +102,7 @@
private DynamicTask dynamicTask;
@Autowired
- private ZlmHttpHookSubscribe subscribe;
+ private CloudRecordServiceMapper cloudRecordServiceMapper;
@Override
@@ -160,10 +155,7 @@
}
}
}
- String streamId = null;
- if (mediaServerItem.isRtpEnable()) {
- streamId = String.format("%s_%s", device.getDeviceId(), channelId);
- }
+ String streamId = String.format("%s_%s", device.getDeviceId(), channelId);;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam());
if (ssrcInfo == null) {
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null);
@@ -459,16 +451,13 @@
logger.warn("[褰曞儚鍥炴斁] 鍗曠鍙ｆ敹娴佹椂涓嶆敮鎸乀CP涓诲姩鏂瑰紡鏀舵祦 deviceId: {},channelId:{}", deviceId, channelId);
throw new ControllerException(ErrorCode.ERROR100.getCode(), "鍗曠鍙ｆ敹娴佹椂涓嶆敮鎸乀CP涓诲姩鏂瑰紡鏀舵祦");
}
- String stream = null;
- if (newMediaServerItem.isRtpEnable()) {
- String startTimeStr = startTime.replace("-", "")
- .replace(":", "")
- .replace(" ", "");
- String endTimeTimeStr = endTime.replace("-", "")
- .replace(":", "")
- .replace(" ", "");
- stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
- }
+ String startTimeStr = startTime.replace("-", "")
+ .replace(":", "")
+ .replace(" ", "");
+ String endTimeTimeStr = endTime.replace("-", "")
+ .replace(":", "")
+ .replace(" ", "");
+ String stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, callback);
}
@@ -630,44 +619,13 @@
if (ssrcInResponse != null) {
// 鍗曠鍙�
// 閲嶆柊璁㈤槄娴佷笂绾�
- HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp",
- ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
- subscribe.removeSubscribe(hookSubscribe);
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), null, inviteInfo.getStream());
streamSession.remove(inviteInfo.getDeviceId(),
inviteInfo.getChannelId(), inviteInfo.getStream());
-
- String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase();
- hookSubscribe.getContent().put("stream", stream);
-
- inviteStreamService.updateInviteInfoForStream(inviteInfo, stream);
+ inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse);
streamSession.put(device.getDeviceId(), channelId, ssrcTransaction.getCallId(),
- stream, ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType);
- subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> {
- logger.info("[Invite 200OK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + hookParam);
- dynamicTask.stop(timeOutTaskKey);
- subscribe.removeSubscribe(hookSubscribe);
- // hook鍝嶅簲
- StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInUse, hookParam, device.getDeviceId(), channelId);
- if (streamInfo == null){
- callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
- InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
- inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null,
- InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
- InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
- return;
- }
- callback.run(InviteErrorCode.SUCCESS.getCode(),
- InviteErrorCode.SUCCESS.getMsg(), streamInfo);
- inviteStreamService.call(inviteSessionType, device.getDeviceId(), channelId, null,
- InviteErrorCode.SUCCESS.getCode(),
- InviteErrorCode.SUCCESS.getMsg(),
- streamInfo);
- if (inviteSessionType == InviteSessionType.PLAY) {
- snapOnPlay(mediaServerItemInUse, device.getDeviceId(), channelId, stream);
- }
- });
+ inviteInfo.getStream(), ssrcInResponse, mediaServerItem.getId(), (SIPResponse) responseEvent.getResponse(), inviteSessionType);
}
}
}
@@ -688,17 +646,8 @@
null);
return;
}
- String stream = null;
- if (newMediaServerItem.isRtpEnable()) {
- String startTimeStr = startTime.replace("-", "")
- .replace(":", "")
- .replace(" ", "");
- String endTimeTimeStr = endTime.replace("-", "")
- .replace(":", "")
- .replace(" ", "");
- stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
- }
- SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
+ // 褰曞儚涓嬭浇涓嶄娇鐢ㄥ浐瀹氭祦鍦板潃锛屽浐瀹氭祦鍦板潃浼氬鑷村鏋滃紑濮嬫椂闂翠笌缁撴潫鏃堕棿涓�鑷存椂鏂囦欢閿欒鐨勫彔鍔犲湪涓�璧�
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, callback);
}
@@ -784,47 +733,147 @@
@Override
public StreamInfo getDownLoadInfo(String deviceId, String channelId, String stream) {
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, stream);
+ if (inviteInfo == null || inviteInfo.getStreamInfo() == null) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 鏈煡璇㈠埌褰曞儚涓嬭浇鐨勪俊鎭�");
+ return null;
+ }
- if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
- if (inviteInfo.getStreamInfo().getProgress() == 1) {
- return inviteInfo.getStreamInfo();
- }
-
- // 鑾峰彇褰撳墠宸蹭笅杞芥椂闀�
- String mediaServerId = inviteInfo.getStreamInfo().getMediaServerId();
- MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
- if (mediaServerItem == null) {
- logger.warn("鏌ヨ褰曞儚淇℃伅鏃跺彂鐜拌妭鐐瑰凡绂荤嚎");
- return null;
- }
- if (mediaServerItem.getRecordAssistPort() > 0) {
- JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, inviteInfo.getStreamInfo().getApp(), inviteInfo.getStreamInfo().getStream(), null);
- if (jsonObject == null) {
- throw new ControllerException(ErrorCode.ERROR100.getCode(), "杩炴帴Assist鏈嶅姟澶辫触");
- }
- if (jsonObject.getInteger("code") == 0) {
- long duration = jsonObject.getLong("data");
-
- if (duration == 0) {
- inviteInfo.getStreamInfo().setProgress(0);
- } else {
- String startTime = inviteInfo.getStreamInfo().getStartTime();
- String endTime = inviteInfo.getStreamInfo().getEndTime();
- long start = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
- long end = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
-
- BigDecimal currentCount = new BigDecimal(duration / 1000);
- BigDecimal totalCount = new BigDecimal(end - start);
- BigDecimal divide = currentCount.divide(totalCount, 2, RoundingMode.HALF_UP);
- double process = divide.doubleValue();
- inviteInfo.getStreamInfo().setProgress(process);
- }
- inviteStreamService.updateInviteInfo(inviteInfo);
- }
- }
+ if (inviteInfo.getStreamInfo().getProgress() == 1) {
return inviteInfo.getStreamInfo();
}
- return null;
+
+ // 鑾峰彇褰撳墠宸蹭笅杞芥椂闀�
+ String mediaServerId = inviteInfo.getStreamInfo().getMediaServerId();
+ MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+ if (mediaServerItem == null) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 鏌ヨ褰曞儚淇℃伅鏃跺彂鐜拌妭鐐逛笉瀛樺湪");
+ return null;
+ }
+ SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, null, stream);
+
+ if (ssrcTransaction == null) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 涓嬭浇宸茬粨鏉�");
+ return null;
+ }
+
+ JSONObject mediaListJson= zlmresTfulUtils.getMediaList(mediaServerItem, "rtp", stream);
+ if (mediaListJson == null) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 浠巣lm鏌ヨ杩涘害澶辫触");
+ return null;
+ }
+ if (mediaListJson.getInteger("code") != 0) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 浠巣lm鏌ヨ杩涘害鍑虹幇閿欒锛� {}", mediaListJson.getString("msg"));
+ return null;
+ }
+ JSONArray data = mediaListJson.getJSONArray("data");
+ if (data == null) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 浠巣lm鏌ヨ杩涘害鏃舵湭杩斿洖鏁版嵁");
+ return null;
+ }
+ JSONObject mediaJSON = data.getJSONObject(0);
+ JSONArray tracks = mediaJSON.getJSONArray("tracks");
+ if (tracks.isEmpty()) {
+ logger.warn("[鑾峰彇涓嬭浇杩涘害] 浠巣lm鏌ヨ杩涘害鏃舵湭杩斿洖鏁版嵁");
+ return null;
+ }
+ JSONObject jsonObject = tracks.getJSONObject(0);
+ long duration = jsonObject.getLongValue("duration");
+ if (duration == 0) {
+ inviteInfo.getStreamInfo().setProgress(0);
+ } else {
+ String startTime = inviteInfo.getStreamInfo().getStartTime();
+ String endTime = inviteInfo.getStreamInfo().getEndTime();
+ // 姝ゆ椂start鍜宔nd鍗曚綅鏄
+ long start = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
+ long end = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
+
+ BigDecimal currentCount = new BigDecimal(duration);
+ BigDecimal totalCount = new BigDecimal((end - start) * 1000);
+ BigDecimal divide = currentCount.divide(totalCount, 2, RoundingMode.HALF_UP);
+ double process = divide.doubleValue();
+ inviteInfo.getStreamInfo().setProgress(process);
+ }
+ inviteStreamService.updateInviteInfo(inviteInfo);
+ return inviteInfo.getStreamInfo();
+ }
+
+ @Override
+ public void getFilePath(String deviceId, String channelId, String stream, ErrorCallback<DownloadFileInfo> callback) {
+ InviteInfo inviteInfo = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, stream);
+ if (inviteInfo == null || inviteInfo.getStreamInfo() == null) {
+ logger.warn("[鑾峰彇褰曞儚涓嬭浇鏂囦欢鍦板潃] 鏈煡璇㈠埌褰曞儚涓嬭浇鐨勪俊鎭紝 {}/{}-{}", deviceId, channelId, stream);
+ callback.run(ErrorCode.ERROR100.getCode(), "鏈煡璇㈠埌褰曞儚涓嬭浇鐨勪俊鎭�", null);
+ return ;
+ }
+
+ if (!ObjectUtils.isEmpty(inviteInfo.getStreamInfo().getDownLoadFilePath())) {
+ callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(),
+ inviteInfo.getStreamInfo().getDownLoadFilePath());
+ return;
+ }
+
+ StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo("rtp", stream);
+ if (streamAuthorityInfo == null) {
+ logger.warn("[鑾峰彇褰曞儚涓嬭浇鏂囦欢鍦板潃] 鏈煡璇㈠埌褰曞儚鐨勮棰戜俊鎭紝 {}/{}-{}", deviceId, channelId, stream);
+ callback.run(ErrorCode.ERROR100.getCode(), "鏈煡璇㈠埌褰曞儚鐨勮棰戜俊鎭�", null);
+ return ;
+ }
+
+ // 鑾峰彇褰撳墠宸蹭笅杞芥椂闀�
+ String mediaServerId = inviteInfo.getStreamInfo().getMediaServerId();
+ MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+ if (mediaServerItem == null) {
+ logger.warn("[鑾峰彇褰曞儚涓嬭浇鏂囦欢鍦板潃] 鏌ヨ褰曞儚淇℃伅鏃跺彂鐜拌妭鐐逛笉瀛樺湪锛� {}/{}-{}", deviceId, channelId, stream);
+ callback.run(ErrorCode.ERROR100.getCode(), "鏌ヨ褰曞儚淇℃伅鏃跺彂鐜拌妭鐐逛笉瀛樺湪", null);
+ return ;
+ }
+
+ List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getListByCallId(streamAuthorityInfo.getCallId());
+ if (!cloudRecordItemList.isEmpty()) {
+ String filePath = cloudRecordItemList.get(0).getFilePath();
+
+ DownloadFileInfo downloadFileInfo = getDownloadFilePath(mediaServerItem, filePath);
+ inviteInfo.getStreamInfo().setDownLoadFilePath(downloadFileInfo);
+ inviteStreamService.updateInviteInfo(inviteInfo);
+ callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), downloadFileInfo);
+ }else {
+ // 鍙兘灏氭湭鐢熸垚锛岄偅灏辩洃鍚琱ook绛夌潃鏀跺埌瀵瑰簲鐨勫綍鍍忛�氱煡
+ ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInuse, hookParam) -> {
+ logger.info("[褰曞儚涓嬭浇]鏀跺埌璁㈤槄娑堟伅锛� 锛� {}/{}-{}", deviceId, channelId, stream);
+ logger.info("[褰曞儚涓嬭浇]鏀跺埌璁㈤槄娑堟伅鍐呭锛� " + hookParam);
+ dynamicTask.stop(streamAuthorityInfo.getCallId());
+ OnRecordMp4HookParam recordMp4HookParam = (OnRecordMp4HookParam)hookParam;
+ String filePath = recordMp4HookParam.getFile_path();
+ DownloadFileInfo downloadFileInfo = getDownloadFilePath(mediaServerItem, filePath);
+ inviteInfo.getStreamInfo().setDownLoadFilePath(downloadFileInfo);
+ inviteStreamService.updateInviteInfo(inviteInfo);
+ callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), downloadFileInfo);
+ };
+ HookSubscribeForRecordMp4 hookSubscribe = HookSubscribeFactory.on_record_mp4(mediaServerId, "rtp", stream);
+ subscribe.addSubscribe(hookSubscribe, hookEvent);
+
+ // 璁剧疆瓒呮椂锛岃秴鏃剁粨鏉熺洃鍚�
+ dynamicTask.startDelay(streamAuthorityInfo.getCallId(), ()->{
+ logger.info("[褰曞儚涓嬭浇] 鎺ユ敹hook瓒呮椂锛� {}/{}-{}", deviceId, channelId, stream);
+ subscribe.removeSubscribe(hookSubscribe);
+ callback.run(ErrorCode.ERROR100.getCode(), "鎺ユ敹hook瓒呮椂", null);
+ }, 10000);
+ }
+ }
+
+ private DownloadFileInfo getDownloadFilePath(MediaServerItem mediaServerItem, String filePath) {
+ DownloadFileInfo downloadFileInfo = new DownloadFileInfo();
+
+ String pathTemplate = "%s://%s:%s/index/api/downloadFile?file_path=" + filePath;
+
+ downloadFileInfo.setHttpPath(String.format(pathTemplate, "http", mediaServerItem.getStreamIp(),
+ mediaServerItem.getHttpPort()));
+
+ if (mediaServerItem.getHttpSSlPort() > 0) {
+ downloadFileInfo.setHttpsPath(String.format(pathTemplate, "https", mediaServerItem.getStreamIp(),
+ mediaServerItem.getHttpSSlPort()));
+ }
+ return downloadFileInfo;
}
private StreamInfo onPublishHandlerForDownload(MediaServerItem mediaServerItemInuse, HookParam hookParam, String deviceId, String channelId, String startTime, String endTime) {
--
Gitblit v1.8.0