Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

基于新的云端录像结构实现国标录像

Browse Source
This commit is contained in:
648540858 2023-10-17 15:34:01 +08:00
parent 7aa8444e67
commit 5a75381a00
14 changed files with 404 additions and 212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java
View File

@ -75,6 +75,33 @@ public class VideoStreamSessionManager {
return (SsrcTransaction)redisTemplate.opsForValue().get(scanResult.get(0));
}
public SsrcTransaction getSsrcTransactionByCallId(String callId){
if (ObjectUtils.isEmpty(callId)) {
return null;
}
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_*_*_" + callId+ "_*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (!scanResult.isEmpty()) {
return (SsrcTransaction)redisTemplate.opsForValue().get(scanResult.get(0));
}else {
key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_*_*_play_*";
scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.isEmpty()) {
return null;
}
for (Object keyObj : scanResult) {
SsrcTransaction ssrcTransaction = (SsrcTransaction)redisTemplate.opsForValue().get(keyObj);
if (ssrcTransaction.getSipTransactionInfo() != null &&
ssrcTransaction.getSipTransactionInfo().getCallId().equals(callId)) {
return ssrcTransaction;
}
}
return null;
}
}
public List<SsrcTransaction> getSsrcTransactionForAll(String deviceId, String channelId, String callId, String stream){
if (ObjectUtils.isEmpty(deviceId)) {
deviceId ="*";

3
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
View File

@ -31,6 +31,7 @@ import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
@ -149,7 +150,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
}else {
// 可能是设备发送的停止
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, callIdHeader.getCallId(), null);
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransactionByCallId(callIdHeader.getCallId());
if (ssrcTransaction == null) {
return;
}

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java
View File

@ -76,7 +76,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress());
if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) {
logger.info("[心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
logger.info("[收到心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
device.setPort(remoteAddressInfo.getPort());
device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
device.setIp(remoteAddressInfo.getIp());

180
src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java
View File

@ -9,12 +9,16 @@ import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import java.io.IOException;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@Component
public class AssistRESTfulUtils {
@ -22,21 +26,43 @@ public class AssistRESTfulUtils {
private final static Logger logger = LoggerFactory.getLogger(AssistRESTfulUtils.class);
private OkHttpClient client;
public interface RequestCallback{
void run(JSONObject response);
}
private OkHttpClient getClient(){
OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
if (logger.isDebugEnabled()) {
HttpLoggingInterceptor logging = new HttpLoggingInterceptor(message -> {
logger.debug("http请求参数：" + message);
});
logging.setLevel(HttpLoggingInterceptor.Level.BASIC);
// OkHttp進行添加攔截器loggingInterceptor
httpClientBuilder.addInterceptor(logging);
return getClient(null);
}
private OkHttpClient getClient(Integer readTimeOut){
if (client == null) {
if (readTimeOut == null) {
readTimeOut = 10;
}
OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
// 设置连接超时时间
httpClientBuilder.connectTimeout(8, TimeUnit.SECONDS);
// 设置读取超时时间
httpClientBuilder.readTimeout(readTimeOut,TimeUnit.SECONDS);
// 设置连接池
httpClientBuilder.connectionPool(new ConnectionPool(16, 5, TimeUnit.MINUTES));
if (logger.isDebugEnabled()) {
HttpLoggingInterceptor logging = new HttpLoggingInterceptor(message -> {
logger.debug("http请求参数：" + message);
});
logging.setLevel(HttpLoggingInterceptor.Level.BASIC);
// OkHttp進行添加攔截器loggingInterceptor
httpClientBuilder.addInterceptor(logging);
}
client = httpClientBuilder.build();
}
return httpClientBuilder.build();
return client;
}
@ -124,13 +150,91 @@ public class AssistRESTfulUtils {
return responseJSON;
}
public JSONObject sendPost(MediaServerItem mediaServerItem, String api, JSONObject param, ZLMRESTfulUtils.RequestCallback callback, Integer readTimeOut) {
OkHttpClient client = getClient(readTimeOut);
public JSONObject fileDuration(MediaServerItem mediaServerItem, String app, String stream, RequestCallback callback){
Map<String, Object> param = new HashMap<>();
param.put("app",app);
param.put("stream",stream);
param.put("recordIng",true);
return sendGet(mediaServerItem, "api/record/file/duration",param, callback);
if (mediaServerItem == null) {
return null;
}
String url = String.format("http://%s:%s/%s", mediaServerItem.getIp(), mediaServerItem.getRecordAssistPort(), api);
JSONObject responseJSON = new JSONObject();
//-2自定义流媒体 调用错误码
responseJSON.put("code",-2);
responseJSON.put("msg","ASSIST调用失败");
RequestBody requestBodyJson = RequestBody.create(MediaType.parse("application/json; charset=utf-8"), param.toString());
Request request = new Request.Builder()
.post(requestBodyJson)
.url(url)
.addHeader("Content-Type", "application/json")
.build();
if (callback == null) {
try {
Response response = client.newCall(request).execute();
if (response.isSuccessful()) {
ResponseBody responseBody = response.body();
if (responseBody != null) {
String responseStr = responseBody.string();
responseJSON = JSON.parseObject(responseStr);
}
}else {
response.close();
Objects.requireNonNull(response.body()).close();
}
}catch (IOException e) {
logger.error(String.format("[ %s ]ASSIST请求失败: %s", url, e.getMessage()));
if(e instanceof SocketTimeoutException){
//读取超时超时异常
logger.error(String.format("读取ASSIST数据失败: %s, %s", url, e.getMessage()));
}
if(e instanceof ConnectException){
//判断连接异常，我这里是报Failed to connect to 10.7.5.144
logger.error(String.format("连接ASSIST失败: %s, %s", url, e.getMessage()));
}
}catch (Exception e){
logger.error(String.format("访问ASSIST失败: %s, %s", url, e.getMessage()));
}
}else {
client.newCall(request).enqueue(new Callback(){
@Override
public void onResponse(@NotNull Call call, @NotNull Response response){
if (response.isSuccessful()) {
try {
String responseStr = Objects.requireNonNull(response.body()).string();
callback.run(JSON.parseObject(responseStr));
} catch (IOException e) {
logger.error(String.format("[ %s ]请求失败: %s", url, e.getMessage()));
}
}else {
response.close();
Objects.requireNonNull(response.body()).close();
}
}
@Override
public void onFailure(@NotNull Call call, @NotNull IOException e) {
logger.error(String.format("连接ZLM失败: %s, %s", call.request().toString(), e.getMessage()));
if(e instanceof SocketTimeoutException){
//读取超时超时异常
logger.error(String.format("读取ZLM数据失败: %s, %s", call.request().toString(), e.getMessage()));
}
if(e instanceof ConnectException){
//判断连接异常，我这里是报Failed to connect to 10.7.5.144
logger.error(String.format("连接ZLM失败: %s, %s", call.request().toString(), e.getMessage()));
}
}
});
}
return responseJSON;
}
public JSONObject getInfo(MediaServerItem mediaServerItem, RequestCallback callback){
@ -138,33 +242,33 @@ public class AssistRESTfulUtils {
return sendGet(mediaServerItem, "api/record/info",param, callback);
}
public JSONObject addStreamCallInfo(MediaServerItem mediaServerItem, String app, String stream, String callId, RequestCallback callback){
Map<String, Object> param = new HashMap<>();
param.put("app",app);
param.put("stream",stream);
param.put("callId",callId);
return sendGet(mediaServerItem, "api/record/addStreamCallInfo",param, callback);
public JSONObject addTask(MediaServerItem mediaServerItem, String app, String stream, String startTime,
String endTime, String callId, List<String> filePathList, String remoteHost) {
JSONObject videoTaskInfoJSON = new JSONObject();
videoTaskInfoJSON.put("app", app);
videoTaskInfoJSON.put("stream", stream);
videoTaskInfoJSON.put("startTime", startTime);
videoTaskInfoJSON.put("endTime", endTime);
videoTaskInfoJSON.put("callId", callId);
videoTaskInfoJSON.put("filePathList", filePathList);
if (!ObjectUtils.isEmpty(remoteHost)) {
videoTaskInfoJSON.put("remoteHost", remoteHost);
}
return sendPost(mediaServerItem, "api/record/file/download/task/add", videoTaskInfoJSON, null, 30);
}
public JSONObject getDateList(MediaServerItem mediaServerItem, String app, String stream, int year, int month) {
public JSONObject queryTaskList(MediaServerItem mediaServerItem, String taskId, Boolean isEnd) {
Map<String, Object> param = new HashMap<>();
param.put("app", app);
param.put("stream", stream);
param.put("year", year);
param.put("month", month);
return sendGet(mediaServerItem, "api/record/date/list", param, null);
}
if (!ObjectUtils.isEmpty(taskId)) {
param.put("taskId", taskId);
}
if (!ObjectUtils.isEmpty(isEnd)) {
param.put("isEnd", isEnd);
}
public JSONObject getFileList(MediaServerItem mediaServerItem, int page, int count, String app, String stream,
String startTime, String endTime) {
Map<String, Object> param = new HashMap<>();
param.put("app", app);
param.put("stream", stream);
param.put("page", page);
param.put("count", count);
param.put("startTime", startTime);
param.put("endTime", endTime);
return sendGet(mediaServerItem, "api/record/file/listWithDate", param, null);
return sendGet(mediaServerItem, "api/record/file/download/task/list", param, null);
}
}

35
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@ -234,12 +234,6 @@ public class ZLMHttpHookListener {
streamAuthorityInfo.setSign(sign);
// 鉴权通过
redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
// 通知assist新的callId
if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {
taskExecutor.execute(() -> {
assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);
});
}
}
} else {
zlmMediaListManager.sendStreamEvent(param.getApp(), param.getStream(), param.getMediaServerId());
@ -267,15 +261,28 @@ public class ZLMHttpHookListener {
}
// 替换流地址
if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) {
String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
if (inviteInfo != null) {
result.setStream_replace(inviteInfo.getStream());
logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
if (!mediaInfo.isRtpEnable()) {
String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
if (inviteInfo != null) {
result.setStream_replace(inviteInfo.getStream());
logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
}
}
}
List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
// 为录制国标模拟一个鉴权信息
StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
streamAuthorityInfo.setApp(param.getApp());
streamAuthorityInfo.setStream(ssrcTransactionForAll.get(0).getStream());
streamAuthorityInfo.setCallId(ssrcTransactionForAll.get(0).getSipTransactionInfo().getCallId());
redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), ssrcTransactionForAll.get(0).getStream(), streamAuthorityInfo);
String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
String channelId = ssrcTransactionForAll.get(0).getChannelId();
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
@ -349,13 +356,11 @@ public class ZLMHttpHookListener {
List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
// TODO 重构此处逻辑
boolean isPush = false;
if (param.isRegist()) {
// 处理流注册的鉴权信息
// 处理流注册的鉴权信息， 流注销这里不再删除鉴权信息，下次来了新的鉴权信息会对就的进行覆盖
if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
isPush = true;
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
if (streamAuthorityInfo == null) {
streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
@ -365,8 +370,6 @@ public class ZLMHttpHookListener {
}
redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
}
} else {
redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());
}
if ("rtsp".equals(param.getSchema())) {

12
src/main/java/com/genersoft/iot/vmp/service/ICloudRecordService.java
View File

@ -1,5 +1,7 @@
package com.genersoft.iot.vmp.service;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
@ -28,4 +30,14 @@ public interface ICloudRecordService {
*/
List<String> getDateList(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);
/**
* 添加合并任务
*/
String addTask(String app, String stream, String mediaServerId, String startTime, String endTime, String callId, String remoteHost);
/**
* 查询合并任务列表
*/
JSONArray queryTask(String taskId, String mediaServerId, Boolean isEnd);
}

11
src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
View File

@ -87,21 +87,10 @@ public interface IMediaServerService {
void updateMediaServerKeepalive(String mediaServerId, ServerKeepaliveData data);
boolean checkRtpServer(MediaServerItem mediaServerItem, String rtp, String stream);
/**
* 获取负载信息
* @return
*/
MediaServerLoad getLoad(MediaServerItem mediaServerItem);
/**
* 按时间查找录像文件
*/
List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems);
/**
* 查找存在录像文件的时间
*/
List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);
}

84
src/main/java/com/genersoft/iot/vmp/service/impl/CloudRecordServiceImpl.java
View File

@ -1,11 +1,17 @@
package com.genersoft.iot.vmp.service.impl;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
import com.genersoft.iot.vmp.service.ICloudRecordService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
@ -31,9 +37,18 @@ public class CloudRecordServiceImpl implements ICloudRecordService {
@Autowired
private CloudRecordServiceMapper cloudRecordServiceMapper;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private AssistRESTfulUtils assistRESTfulUtils;
@Autowired
private VideoStreamSessionManager streamSession;
@Override
public PageInfo<CloudRecordItem> getList(int page, int count, String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
// 开始时间和结束时间在数据库中都是以秒为单位的
@ -54,7 +69,8 @@ public class CloudRecordServiceImpl implements ICloudRecordService {
}
PageHelper.startPage(page, count);
List<CloudRecordItem> all = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp, mediaServerItems);
List<CloudRecordItem> all = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp,
null, mediaServerItems);
return new PageInfo<>(all);
}
@ -69,7 +85,8 @@ public class CloudRecordServiceImpl implements ICloudRecordService {
}
long startTimeStamp = startDate.atStartOfDay().toInstant(ZoneOffset.ofHours(8)).getEpochSecond();
long endTimeStamp = endDate.atStartOfDay().toInstant(ZoneOffset.ofHours(8)).getEpochSecond();
List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp, mediaServerItems);
List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList(app, stream, startTimeStamp,
endTimeStamp, null, mediaServerItems);
if (cloudRecordItemList.isEmpty()) {
return new ArrayList<>();
}
@ -83,12 +100,71 @@ public class CloudRecordServiceImpl implements ICloudRecordService {
@Override
public void addRecord(OnRecordMp4HookParam param) {
CloudRecordItem cloudRecordItem = CloudRecordItem.getInstance(param);
CloudRecordItem cloudRecordItem = CloudRecordItem.getInstance(param);
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
if (streamAuthorityInfo != null) {
cloudRecordItem.setCallId(streamAuthorityInfo.getCallId());
}
logger.info("[添加录像记录] {}/{} 文件大小：{}", param.getApp(), param.getStream(), param.getFile_size());
logger.info("[添加录像记录] {}/{} 文件大小：{}, 时长： {}秒", param.getApp(), param.getStream(), param.getFile_size(),param.getTime_len());
cloudRecordServiceMapper.add(cloudRecordItem);
}
@Override
public String addTask(String app, String stream, String mediaServerId, String startTime, String endTime, String callId, String remoteHost) {
// 参数校验
assert app != null;
assert stream != null;
MediaServerItem mediaServerItem = null;
if (mediaServerId == null) {
mediaServerItem = mediaServerService.getDefaultMediaServer();
}else {
mediaServerItem = mediaServerService.getOne(mediaServerId);
}
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的流媒体");
}else {
if (remoteHost == null) {
remoteHost = "http://" + mediaServerItem.getStreamIp() + ":" + mediaServerItem.getRecordAssistPort();
}
}
Long startTimeStamp = null;
Long endTimeStamp = null;
if (startTime != null) {
startTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
}
if (endTime != null) {
endTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
}
List<MediaServerItem> mediaServers = new ArrayList<>();
mediaServers.add(mediaServerItem);
// 检索相关的录像文件
List<String> filePathList = cloudRecordServiceMapper.queryRecordFilePathList(app, stream, startTimeStamp, endTimeStamp, callId, mediaServers);
if (filePathList == null || filePathList.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未检索到视频文件");
}
JSONObject result = assistRESTfulUtils.addTask(mediaServerItem, app, stream, startTime, endTime, callId, filePathList, remoteHost);
if (result.getInteger("code") != 0) {
throw new ControllerException(result.getInteger("code"), result.getString("msg"));
}
return result.getString("data");
}
@Override
public JSONArray queryTask(String taskId, String mediaServerId, Boolean isEnd) {
MediaServerItem mediaServerItem = null;
if (mediaServerId == null) {
mediaServerItem = mediaServerService.getDefaultMediaServer();
}else {
mediaServerItem = mediaServerService.getOne(mediaServerId);
}
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的流媒体");
}
JSONObject result = assistRESTfulUtils.queryTaskList(mediaServerItem, taskId, isEnd);
if (result.getInteger("code") != 0) {
throw new ControllerException(result.getInteger("code"), result.getString("msg"));
}
return result.getJSONArray("data");
}
}

95
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
View File

@ -741,15 +741,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
}
@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();
@ -761,90 +752,4 @@ public class MediaServerServiceImpl implements IMediaServerService {
result.setGbSend(redisCatchStorage.getGbSendCount(mediaServerItem.getId()));
return result;
}
@Override
public List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
Assert.notNull(app, "app不存在");
Assert.notNull(stream, "stream不存在");
Assert.notNull(startTime, "startTime不存在");
Assert.notNull(endTime, "endTime不存在");
Assert.notEmpty(mediaServerItems, "流媒体列表为空");
CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];
for (int i = 0; i < mediaServerItems.size(); i++) {
completableFutures[i] = getRecordFilesForOne(app, stream, startTime, endTime, mediaServerItems.get(i));
}
List<RecordFile> result = new ArrayList<>();
for (int i = 0; i < completableFutures.length; i++) {
try {
List<RecordFile> list = (List<RecordFile>) completableFutures[i].get();
if (!list.isEmpty()) {
for (int g = 0; g < list.size(); g++) {
list.get(g).setMediaServerId(mediaServerItems.get(i).getId());
}
result.addAll(list);
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
}
Comparator<RecordFile> comparator = Comparator.comparing(RecordFile::getFileName);
result.sort(comparator);
return result;
}
@Override
public List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems) {
Assert.notNull(app, "app不存在");
Assert.notNull(stream, "stream不存在");
Assert.notEmpty(mediaServerItems, "流媒体列表为空");
CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];
for (int i = 0; i < mediaServerItems.size(); i++) {
completableFutures[i] = getRecordDatesForOne(app, stream, year, month, mediaServerItems.get(i));
}
List<String> result = new ArrayList<>();
CompletableFuture.allOf(completableFutures).join();
for (CompletableFuture completableFuture : completableFutures) {
try {
List<String> list = (List<String>) completableFuture.get();
result.addAll(list);
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
}
Collections.sort(result);
return result;
}
@Async
public CompletableFuture<List<String>> getRecordDatesForOne(String app, String stream, int year, int month, MediaServerItem mediaServerItem) {
JSONObject fileListJson = assistRESTfulUtils.getDateList(mediaServerItem, app, stream, year, month);
if (fileListJson != null && !fileListJson.isEmpty()) {
if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
JSONArray data = fileListJson.getJSONArray("data");
return CompletableFuture.completedFuture(data.toJavaList(String.class));
}
}
return CompletableFuture.completedFuture(new ArrayList<>());
}
@Async
public CompletableFuture<List<RecordFile>> getRecordFilesForOne(String app, String stream, String startTime, String endTime, MediaServerItem mediaServerItem) {
JSONObject fileListJson = assistRESTfulUtils.getFileList(mediaServerItem, 1, 100000000, app, stream, startTime, endTime);
if (fileListJson != null && !fileListJson.isEmpty()) {
if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
JSONObject data = fileListJson.getJSONObject("data");
JSONArray list = data.getJSONArray("list");
if (list != null) {
return CompletableFuture.completedFuture(list.toJavaList(RecordFile.class));
}
}
}
return CompletableFuture.completedFuture(new ArrayList<>());
}
}

65
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@ -27,11 +27,13 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
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.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;
@ -107,6 +109,9 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private ZlmHttpHookSubscribe subscribe;
@Autowired
private CloudRecordServiceMapper cloudRecordServiceMapper;
@Override
public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, String ssrc, ErrorCallback<Object> callback) {
@ -749,31 +754,43 @@ public class PlayServiceImpl implements IPlayService {
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 (mediaServerItem.getRecordAssistPort() == 0) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未配置Assist服务，无法完成录像下载");
}
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) {
logger.warn("[获取下载进度]，未找到下载事务信息");
return null;
}
// 为了支持多个数据库，这里不能使用求和函数来直接获取总数了
List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList("rtp", inviteInfo.getStream(), null, null, ssrcTransaction.getCallId(), null);
if (cloudRecordItemList.isEmpty()) {
logger.warn("[获取下载进度]，未找到下载视频信息");
return null;
}
long duration = 0;
for (CloudRecordItem cloudRecordItem : cloudRecordItemList) {
duration += cloudRecordItem.getTimeLen();
}
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);
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);
return inviteInfo.getStreamInfo();
}
return null;

30
src/main/java/com/genersoft/iot/vmp/storager/dao/CloudRecordServiceMapper.java
View File

@ -42,17 +42,37 @@ public interface CloudRecordServiceMapper {
@Select(" <script>" +
"select * " +
"from wvp_cloud_record " +
"where 0 = 0" +
" from wvp_cloud_record " +
" where 0 = 0" +
" <if test= 'app != null '> and app=#{app}</if>" +
" <if test= 'stream != null '> and stream=#{stream}</if>" +
" <if test= 'startTimeStamp != null '> and start_time &gt;= #{startTimeStamp}</if>" +
" <if test= 'endTimeStamp != null '> and end_time &lt;= #{endTimeStamp}</if>" +
" <if test= 'startTimeStamp != null '> and end_time &gt;= #{startTimeStamp}</if>" +
" <if test= 'endTimeStamp != null '> and start_time &lt;= #{endTimeStamp}</if>" +
" <if test= 'callId != null '> and call_id = #{callId}</if>" +
" <if test= 'mediaServerItemList != null ' > and media_server_id in " +
" <foreach collection='mediaServerItemList' item='item' open='(' separator=',' close=')' > #{item.id}</foreach>" +
" </if>" +
" </script>")
List<CloudRecordItem> getList(@Param("app") String app, @Param("stream") String stream,
@Param("startTimeStamp")Long startTimeStamp, @Param("endTimeStamp")Long endTimeStamp,
List<MediaServerItem> mediaServerItemList);
@Param("callId")String callId, List<MediaServerItem> mediaServerItemList);
@Select(" <script>" +
"select file_path" +
" from wvp_cloud_record " +
" where 0 = 0" +
" <if test= 'app != null '> and app=#{app}</if>" +
" <if test= 'stream != null '> and stream=#{stream}</if>" +
" <if test= 'startTimeStamp != null '> and end_time &gt;= #{startTimeStamp}</if>" +
" <if test= 'endTimeStamp != null '> and start_time &lt;= #{endTimeStamp}</if>" +
" <if test= 'callId != null '> and call_id = #{callId}</if>" +
" <if test= 'mediaServerItemList != null ' > and media_server_id in " +
" <foreach collection='mediaServerItemList' item='item' open='(' separator=',' close=')' > #{item.id}</foreach>" +
" </if>" +
" </script>")
List<String> queryRecordFilePathList(@Param("app") String app, @Param("stream") String stream,
@Param("startTimeStamp")Long startTimeStamp, @Param("endTimeStamp")Long endTimeStamp,
@Param("callId")String callId, List<MediaServerItem> mediaServerItemList);
}

50
src/main/java/com/genersoft/iot/vmp/vmanager/cloudRecord/CloudRecordController.java
View File

@ -1,5 +1,7 @@
package com.genersoft.iot.vmp.vmanager.cloudRecord;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
@ -145,19 +147,53 @@ public class CloudRecordController {
@Operation(summary = "添加合并任务")
@Parameter(name = "app", description = "应用名", required = true)
@Parameter(name = "stream", description = "流ID", required = true)
@Parameter(name = "mediaServerId", description = "流媒体ID", required = false)
@Parameter(name = "startTime", description = "鉴权ID", required = false)
@Parameter(name = "endTime", description = "鉴权ID", required = false)
@Parameter(name = "callId", description = "鉴权ID", required = false)
@Parameter(name = "remoteHost", description = "返回地址时的远程地址", required = false)
public String addTask(
@RequestParam String app,
@RequestParam String stream,
@RequestParam String startTime,
@RequestParam String endTime,
@RequestParam String callId,
@RequestParam String remoteHost
@RequestParam(required = true) String app,
@RequestParam(required = true) String stream,
@RequestParam(required = false) String mediaServerId,
@RequestParam(required = false) String startTime,
@RequestParam(required = false) String endTime,
@RequestParam(required = false) String callId,
@RequestParam(required = false) String remoteHost
){
return cloudRecordService.addTask(app, stream, startTime, endTime, callId, remoteHost);
return cloudRecordService.addTask(app, stream, mediaServerId, startTime, endTime, callId, remoteHost);
}
@ResponseBody
@GetMapping("/task/list")
@Operation(summary = "查询合并任务")
@Parameter(name = "taskId", description = "任务Id", required = false)
@Parameter(name = "mediaServerId", description = "流媒体ID", required = false)
@Parameter(name = "isEnd", description = "是否结束", required = false)
public JSONArray queryTaskList(
@RequestParam(required = false) String taskId,
@RequestParam(required = false) String mediaServerId,
@RequestParam(required = false) Boolean isEnd
){
return cloudRecordService.queryTask(taskId, mediaServerId, isEnd);
}
@ResponseBody
@GetMapping("/collect/add")
@Operation(summary = "添加收藏")
@Parameter(name = "app", description = "应用名", required = true)
@Parameter(name = "stream", description = "流ID", required = true)
@Parameter(name = "mediaServerId", description = "流媒体ID", required = false)
@Parameter(name = "startTime", description = "鉴权ID", required = false)
@Parameter(name = "endTime", description = "鉴权ID", required = false)
@Parameter(name = "callId", description = "鉴权ID", required = false)
@Parameter(name = "collectType", description = "收藏类型", required = false)
public JSONArray addCollect(
@RequestParam(required = false) String taskId,
@RequestParam(required = false) String mediaServerId,
@RequestParam(required = false) Boolean isEnd
){
return cloudRecordService.queryTask(taskId, mediaServerId, isEnd);
}

14
web_src/src/components/CloudRecordDetail.vue
View File

@ -480,12 +480,13 @@
let that = this;
this.$axios({
method: 'get',
url:`/record_proxy/${that.mediaServerId}/api/record/file/download/task/add`,
url:`/api/cloud/record/task/add`,
params: {
app: that.app,
stream: that.stream,
startTime: moment(this.taskTimeRange[0]).format('YYYY-MM-DD HH:mm:ss'),
endTime: moment(this.taskTimeRange[1]).format('YYYY-MM-DD HH:mm:ss'),
app: this.app,
stream: this.stream,
mediaServerId: this.mediaServerId,
startTime: moment(this.taskTimeRange[0]).format('YYYY-MM-DD HH:mm:ss'),
endTime: moment(this.taskTimeRange[1]).format('YYYY-MM-DD HH:mm:ss'),
}
}).then(function (res) {
if (res.data.code === 0 ) {
@ -505,8 +506,9 @@
let that = this;
this.$axios({
method: 'get',
url:`/record_proxy/${that.mediaServerId}/api/record/file/download/task/list`,
url:`/api/cloud/record/task/list`,
params: {
mediaServerId: this.mediaServerId,
isEnd: isEnd,
}
}).then(function (res) {

8
web_src/src/components/dialog/recordDownload.vue
View File

@ -137,10 +137,11 @@ export default {
getFileDownload: function (){
this.$axios({
method: 'get',
url:`/record_proxy/${this.mediaServerId}/api/record/file/download/task/add`,
url:`/api/cloud/record/task/add`,
params: {
app: this.app,
stream: this.stream,
mediaServerId: this.mediaServerId,
startTime: null,
endTime: null,
}
@ -169,10 +170,9 @@ export default {
getProgressForFile: function (callback){
this.$axios({
method: 'get',
url:`/record_proxy/${this.mediaServerId}/api/record/file/download/task/list`,
url:`/api/cloud/record/task/list`,
params: {
app: this.app,
stream: this.stream,
mediaServerId: this.mediaServerId,
taskId: this.taskId,
isEnd: true,
}