XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'wvp-28181-2.0' into main-dev

Browse Source 
# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeFactory.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java
This commit is contained in:
648540858
2024-01-16 14:10:27 +08:00
parent c32ba1dea8 674ab18c02
commit 9a96597e66
116 changed files with 9493 additions and 16489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
src/main/java/com/genersoft/iot/vmp/service/ICloudRecordService.java Executable file
View File

@@ -0,0 +1,58 @@
package com.genersoft.iot.vmp.service;
import com.alibaba.fastjson2.JSONArray;
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;
import com.genersoft.iot.vmp.service.bean.DownloadFileInfo;
import com.github.pagehelper.PageInfo;
import java.util.List;
/**
* 云端录像管理
* @author lin
*/
public interface ICloudRecordService {
/**
* 分页回去云端录像列表
*/
PageInfo<CloudRecordItem> getList(int page, int count, String query, String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems);
/**
* 根据hook消息增加一条记录
*/
void addRecord(OnRecordMp4HookParam param);
/**
* 获取所有的日期
*/
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 app, String stream, String callId, String taskId, String mediaServerId, Boolean isEnd);
/**
* 收藏视频，收藏的视频过期不会删除
*/
int changeCollect(boolean result, String app, String stream, String mediaServerId, String startTime, String endTime, String callId);
/**
* 添加指定录像收藏
*/
int changeCollectById(Integer recordId, boolean result);
/**
* 获取播放地址
*/
DownloadFileInfo getPlayUrlPath(Integer recordId);
}

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

@@ -89,21 +89,12 @@ 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<MediaServerItem> getAllWithAssistPort();
/**
* 查找存在录像文件的时间
*/
List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);
}

6
src/main/java/com/genersoft/iot/vmp/service/IPlayService.java
View File

@@ -33,11 +33,6 @@ public interface IPlayService {
MediaServerItem getNewMediaServerItem(Device device);
/**
* 获取包含assist服务的节点
*/
MediaServerItem getNewMediaServerItemHasAssist(Device device);
void playBack(String deviceId, String channelId, String startTime, String endTime, ErrorCallback<Object> callback);
void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, ErrorCallback<Object> callback);
void zlmServerOffline(String mediaServerId);
@@ -72,5 +67,4 @@ public interface IPlayService {
void getSnap(String deviceId, String channelId, String fileName, ErrorCallback errorCallback);
}

1
src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
View File

@@ -114,4 +114,5 @@ public interface IStreamPushService {
* @return
*/
ResourceBaseInfo getOverview();
}

205
src/main/java/com/genersoft/iot/vmp/service/bean/CloudRecordItem.java Normal file
View File

@@ -0,0 +1,205 @@
package com.genersoft.iot.vmp.service.bean;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
/**
* 云端录像数据
*/
public class CloudRecordItem {
/**
* 主键
*/
private int id;
/**
* 应用名
*/
private String app;
/**
* 流
*/
private String stream;
/**
* 健全ID
*/
private String callId;
/**
* 开始时间
*/
private long startTime;
/**
* 结束时间
*/
private long endTime;
/**
* ZLM Id
*/
private String mediaServerId;
/**
* 文件名称
*/
private String fileName;
/**
* 文件路径
*/
private String filePath;
/**
* 文件夹
*/
private String folder;
/**
* 收藏，收藏的文件不移除
*/
private Boolean collect;
/**
* 保留，收藏的文件不移除
*/
private Boolean reserve;
/**
* 文件大小
*/
private long fileSize;
/**
* 文件时长
*/
private long timeLen;
public static CloudRecordItem getInstance(OnRecordMp4HookParam param) {
CloudRecordItem cloudRecordItem = new CloudRecordItem();
cloudRecordItem.setApp(param.getApp());
cloudRecordItem.setStream(param.getStream());
cloudRecordItem.setStartTime(param.getStart_time()*1000);
cloudRecordItem.setFileName(param.getFile_name());
cloudRecordItem.setFolder(param.getFolder());
cloudRecordItem.setFileSize(param.getFile_size());
cloudRecordItem.setFilePath(param.getFile_path());
cloudRecordItem.setMediaServerId(param.getMediaServerId());
cloudRecordItem.setTimeLen((long) param.getTime_len() * 1000);
cloudRecordItem.setEndTime((param.getStart_time() + (long)param.getTime_len()) * 1000);
return cloudRecordItem;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getApp() {
return app;
}
public void setApp(String app) {
this.app = app;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
public String getCallId() {
return callId;
}
public void setCallId(String callId) {
this.callId = callId;
}
public long getStartTime() {
return startTime;
}
public void setStartTime(long startTime) {
this.startTime = startTime;
}
public long getEndTime() {
return endTime;
}
public void setEndTime(long endTime) {
this.endTime = endTime;
}
public String getMediaServerId() {
return mediaServerId;
}
public void setMediaServerId(String mediaServerId) {
this.mediaServerId = mediaServerId;
}
public String getFileName() {
return fileName;
}
public void setFileName(String fileName) {
this.fileName = fileName;
}
public String getFilePath() {
return filePath;
}
public void setFilePath(String filePath) {
this.filePath = filePath;
}
public String getFolder() {
return folder;
}
public void setFolder(String folder) {
this.folder = folder;
}
public long getFileSize() {
return fileSize;
}
public void setFileSize(long fileSize) {
this.fileSize = fileSize;
}
public long getTimeLen() {
return timeLen;
}
public void setTimeLen(long timeLen) {
this.timeLen = timeLen;
}
public Boolean getCollect() {
return collect;
}
public void setCollect(Boolean collect) {
this.collect = collect;
}
public Boolean getReserve() {
return reserve;
}
public void setReserve(Boolean reserve) {
this.reserve = reserve;
}
}

41
src/main/java/com/genersoft/iot/vmp/service/bean/DownloadFileInfo.java Normal file
View File

@@ -0,0 +1,41 @@
package com.genersoft.iot.vmp.service.bean;
public class DownloadFileInfo {
private String httpPath;
private String httpsPath;
private String httpDomainPath;
private String httpsDomainPath;
public String getHttpPath() {
return httpPath;
}
public void setHttpPath(String httpPath) {
this.httpPath = httpPath;
}
public String getHttpsPath() {
return httpsPath;
}
public void setHttpsPath(String httpsPath) {
this.httpsPath = httpsPath;
}
public String getHttpDomainPath() {
return httpDomainPath;
}
public void setHttpDomainPath(String httpDomainPath) {
this.httpDomainPath = httpDomainPath;
}
public String getHttpsDomainPath() {
return httpsDomainPath;
}
public void setHttpsDomainPath(String httpsDomainPath) {
this.httpsDomainPath = httpsDomainPath;
}
}

10
src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsg.java
View File

@@ -29,12 +29,12 @@ public class WvpRedisMsg {
* 消息的ID
*/
private String serial;
private Object content;
private String content;
private final static String requestTag = "req";
private final static String responseTag = "res";
public static WvpRedisMsg getRequestInstance(String fromId, String toId, String cmd, String serial, Object content) {
public static WvpRedisMsg getRequestInstance(String fromId, String toId, String cmd, String serial, String content) {
WvpRedisMsg wvpRedisMsg = new WvpRedisMsg();
wvpRedisMsg.setType(requestTag);
wvpRedisMsg.setFromId(fromId);
@@ -51,7 +51,7 @@ public class WvpRedisMsg {
return wvpRedisMsg;
}
public static WvpRedisMsg getResponseInstance(String fromId, String toId, String cmd, String serial, Object content) {
public static WvpRedisMsg getResponseInstance(String fromId, String toId, String cmd, String serial, String content) {
WvpRedisMsg wvpRedisMsg = new WvpRedisMsg();
wvpRedisMsg.setType(responseTag);
wvpRedisMsg.setFromId(fromId);
@@ -106,11 +106,11 @@ public class WvpRedisMsg {
this.cmd = cmd;
}
public Object getContent() {
public String getContent() {
return content;
}
public void setContent(Object content) {
public void setContent(String content) {
this.content = content;
}
}

242
src/main/java/com/genersoft/iot/vmp/service/impl/CloudRecordServiceImpl.java Normal file
View File

@@ -0,0 +1,242 @@
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.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.service.bean.DownloadFileInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
import com.genersoft.iot.vmp.utils.CloudRecordUtils;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.time.*;
import java.util.*;
@Service
public class CloudRecordServiceImpl implements ICloudRecordService {
private final static Logger logger = LoggerFactory.getLogger(CloudRecordServiceImpl.class);
@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 query, String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
// 开始时间和结束时间在数据库中都是以秒为单位的
Long startTimeStamp = null;
Long endTimeStamp = null;
if (startTime != null ) {
if (!DateUtil.verification(startTime, DateUtil.formatter)) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "开始时间格式错误，正确格式为： " + DateUtil.formatter);
}
startTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
}
if (endTime != null ) {
if (!DateUtil.verification(endTime, DateUtil.formatter)) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "结束时间格式错误，正确格式为： " + DateUtil.formatter);
}
endTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
}
PageHelper.startPage(page, count);
List<CloudRecordItem> all = cloudRecordServiceMapper.getList(query, app, stream, startTimeStamp, endTimeStamp,
null, mediaServerItems);
return new PageInfo<>(all);
}
@Override
public List<String> getDateList(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems) {
LocalDate startDate = LocalDate.of(year, month, 1);
LocalDate endDate;
if (month == 12) {
endDate = LocalDate.of(year + 1, 1, 1);
}else {
endDate = LocalDate.of(year, month + 1, 1);
}
long startTimeStamp = startDate.atStartOfDay().toInstant(ZoneOffset.ofHours(8)).getEpochSecond();
long endTimeStamp = endDate.atStartOfDay().toInstant(ZoneOffset.ofHours(8)).getEpochSecond();
List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList(null, app, stream, startTimeStamp,
endTimeStamp, null, mediaServerItems);
if (cloudRecordItemList.isEmpty()) {
return new ArrayList<>();
}
Set<String> resultSet = new HashSet<>();
cloudRecordItemList.stream().forEach(cloudRecordItem -> {
String date = DateUtil.timestampTo_yyyy_MM_dd(cloudRecordItem.getStartTime());
resultSet.add(date);
});
return new ArrayList<>(resultSet);
}
@Override
public void addRecord(OnRecordMp4HookParam 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(),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();
}
}
if (mediaServerItem.getRecordAssistPort() == 0) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "为配置Assist服务");
}
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 app, String stream, String callId, 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, app, stream, callId, taskId, isEnd);
if (result.getInteger("code") != 0) {
throw new ControllerException(result.getInteger("code"), result.getString("msg"));
}
return result.getJSONArray("data");
}
@Override
public int changeCollect(boolean result, String app, String stream, String mediaServerId, String startTime, String endTime, String callId) {
// 开始时间和结束时间在数据库中都是以秒为单位的
Long startTimeStamp = null;
Long endTimeStamp = null;
if (startTime != null ) {
if (!DateUtil.verification(startTime, DateUtil.formatter)) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "开始时间格式错误，正确格式为： " + DateUtil.formatter);
}
startTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
}
if (endTime != null ) {
if (!DateUtil.verification(endTime, DateUtil.formatter)) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "结束时间格式错误，正确格式为： " + DateUtil.formatter);
}
endTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
}
List<MediaServerItem> mediaServerItems;
if (!ObjectUtils.isEmpty(mediaServerId)) {
mediaServerItems = new ArrayList<>();
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到流媒体: " + mediaServerId);
}
mediaServerItems.add(mediaServerItem);
} else {
mediaServerItems = null;
}
List<CloudRecordItem> all = cloudRecordServiceMapper.getList(null, app, stream, startTimeStamp, endTimeStamp,
callId, mediaServerItems);
if (all.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到待收藏的视频");
}
int limitCount = 50;
int resultCount = 0;
if (all.size() > limitCount) {
for (int i = 0; i < all.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > all.size()) {
toIndex = all.size();
}
resultCount += cloudRecordServiceMapper.updateCollectList(result, all.subList(i, toIndex));
}
}else {
resultCount = cloudRecordServiceMapper.updateCollectList(result, all);
}
return resultCount;
}
@Override
public int changeCollectById(Integer recordId, boolean result) {
return cloudRecordServiceMapper.changeCollectById(result, recordId);
}
@Override
public DownloadFileInfo getPlayUrlPath(Integer recordId) {
CloudRecordItem recordItem = cloudRecordServiceMapper.queryOne(recordId);
if (recordItem == null) {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "资源不存在");
}
String filePath = recordItem.getFilePath();
MediaServerItem mediaServerItem = mediaServerService.getOne(recordItem.getMediaServerId());
return CloudRecordUtils.getDownloadFilePath(mediaServerItem, filePath);
}
}

34
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
View File

@@ -162,6 +162,19 @@ public class DeviceServiceImpl implements IDeviceService {
sync(device);
// TODO 如果设备下的通道级联到了其他平台，那么需要发送事件或者notify给上级平台
}
// 上线添加订阅
if (device.getSubscribeCycleForCatalog() > 0) {
// 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
addCatalogSubscribe(device);
}
if (device.getSubscribeCycleForMobilePosition() > 0) {
addMobilePositionSubscribe(device);
}
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
}
}else {
if (deviceChannelMapper.queryAllChannels(device.getDeviceId()).size() == 0) {
logger.info("[设备上线]: {}，通道数为0,查询通道信息", device.getDeviceId());
@@ -174,22 +187,10 @@ public class DeviceServiceImpl implements IDeviceService {
}
// 上线添加订阅
if (device.getSubscribeCycleForCatalog() > 0) {
// 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
addCatalogSubscribe(device);
}
if (device.getSubscribeCycleForMobilePosition() > 0) {
addMobilePositionSubscribe(device);
}
// 刷新过期任务
String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + device.getDeviceId();
// 如果第一次注册那么必须在60 * 3时间内收到一个心跳，否则设备离线
dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId(), "首次注册后未能收到心跳"), device.getKeepaliveIntervalTime() * 1000 * 3);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
}
//
// try {
@@ -213,6 +214,13 @@ public class DeviceServiceImpl implements IDeviceService {
}
String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + deviceId;
dynamicTask.stop(registerExpireTaskKey);
if (device.isOnLine()) {
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, false);
}
}
device.setOnLine(false);
redisCatchStorage.updateDevice(device);
deviceMapper.update(device);
@@ -224,7 +232,7 @@ public class DeviceServiceImpl implements IDeviceService {
for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
streamSession.remove(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
streamSession.removeByCallId(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
}
}
// 移除订阅

3
src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
View File

@@ -250,9 +250,6 @@ public class GbStreamServiceImpl implements IGbStreamService {
if (platform == null) {
return ;
}
if (ObjectUtils.isEmpty(catalogId)) {
catalogId = platform.getDeviceGBId();
}
if (platformGbStreamMapper.delByPlatformAndCatalogId(platformId, catalogId) > 0) {
List<GbStream> gbStreams = platformGbStreamMapper.queryChannelInParentPlatformAndCatalog(platformId, catalogId);
List<DeviceChannel> deviceChannelList = new ArrayList<>();

5
src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java
View File

@@ -116,9 +116,12 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
":" + (stream != null ? stream : "*")
+ ":*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() != 1) {
if (scanResult.isEmpty()) {
return null;
}
if (scanResult.size() != 1) {
logger.warn("[获取InviteInfo] 发现 key: {}存在多条", key);
}
return (InviteInfo) redisTemplate.opsForValue().get(scanResult.get(0));
}

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

@@ -165,14 +165,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (streamId == null) {
streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase();
}
int ssrcCheckParam = 0;
if (ssrcCheck && tcpMode > 1) {
if (ssrcCheck && tcpMode > 0) {
// 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验
logger.warn("[openRTPServer] TCP被动/TCP主动收流时，默认关闭ssrc检验");
logger.warn("[openRTPServer] 平台对接时下级可能自定义ssrc，但是tcp模式zlm收流目前无法更新ssrc，可能收流超时，此时请使用udp收流或者关闭ssrc校验");
}
int rtpServerPort;
if (mediaServerItem.isRtpEnable()) {
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0) ? Long.parseLong(ssrc) : 0, port, onlyAuto, reUsePort, tcpMode);
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck ? Long.parseLong(ssrc) : 0, port, onlyAuto, reUsePort, tcpMode);
} else {
rtpServerPort = mediaServerItem.getRtpProxyPort();
}
@@ -205,7 +204,10 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public void closeRTPServer(String mediaServerId, String streamId) {
MediaServerItem mediaServerItem = this.getOne(mediaServerId);
closeRTPServer(mediaServerItem, streamId);
if (mediaServerItem.isRtpEnable()) {
closeRTPServer(mediaServerItem, streamId);
}
zlmresTfulUtils.closeStreams(mediaServerItem, "rtp", streamId);
}
@Override
@@ -428,17 +430,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (serverItem.isAutoConfig()) {
// 查看assist服务的录像路径配置
if (serverItem.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {
JSONObject info = assistRESTfulUtils.getInfo(serverItem, null);
if (info != null && info.getInteger("code") != null && info.getInteger("code") == 0 ) {
JSONObject dataJson = info.getJSONObject("data");
if (dataJson != null) {
String recordPath = dataJson.getString("record");
userSetting.setRecordPath(recordPath);
}
}
}
setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));
}
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId();
@@ -573,7 +564,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
logger.info("[ZLM] 正在设置 ：{} -> {}:{}",
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
String protocol = sslEnabled ? "https" : "http";
String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
String hookPrefix = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
@@ -582,25 +573,21 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
param.put("hook.enable","1");
param.put("hook.on_flow_report","");
param.put("hook.on_play",String.format("%s/on_play", hookPrex));
param.put("hook.on_play",String.format("%s/on_play", hookPrefix));
param.put("hook.on_http_access","");
param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
param.put("hook.on_publish", String.format("%s/on_publish", hookPrefix));
param.put("hook.on_record_ts","");
param.put("hook.on_rtsp_auth","");
param.put("hook.on_rtsp_realm","");
param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrex));
param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrefix));
param.put("hook.on_shell_login","");
param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex));
param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrex));
if (mediaServerItem.getRecordAssistPort() > 0) {
param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
}else {
param.put("hook.on_record_mp4","");
}
param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrefix));
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrefix));
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrefix));
param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrefix));
param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrefix));
param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrefix));
param.put("hook.on_record_mp4",String.format("%s/on_record_mp4", hookPrefix));
param.put("hook.timeoutSec","20");
// 推流断开后可以在超时时间内重新连接上继续推流，这样播放器会接着播放。
// 置0关闭此特性(推流断开会导致立即断开播放器)
@@ -609,15 +596,14 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("protocol.continue_push_ms", "3000" );
// 最多等待未初始化的Track时间，单位毫秒，超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流，
// 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
// param.put("general.wait_track_ready_ms", "3000" );
if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
}
if (userSetting.getRecordPath() != null) {
File recordPathFile = new File(userSetting.getRecordPath());
File mp4SavePathFile = recordPathFile.getParentFile().getAbsoluteFile();
param.put("protocol.mp4_save_path", mp4SavePathFile.getAbsoluteFile());
if (!ObjectUtils.isEmpty(mediaServerItem.getRecordPath())) {
File recordPathFile = new File(mediaServerItem.getRecordPath());
param.put("protocol.mp4_save_path", recordPathFile.getParentFile().getPath());
param.put("protocol.downloadRoot", recordPathFile.getParentFile().getPath());
param.put("record.appName", recordPathFile.getName());
}
@@ -722,6 +708,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null);
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
redisTemplate.opsForValue().set(key, mediaServerItem);
resetOnlineServerItem(mediaServerItem);
clearRTPServer(mediaServerItem);
}
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId();
@@ -749,15 +736,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();
@@ -771,88 +749,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
@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<>());
public List<MediaServerItem> getAllWithAssistPort() {
return mediaServerMapper.queryAllWithAssistPort();
}
}

2
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
View File

@@ -64,7 +64,7 @@ public class MediaServiceImpl implements IMediaService {
if (data == null) {
return null;
}
JSONObject mediaJSON = JSON.parseObject(JSON.toJSONString(data.get(0)), JSONObject.class);
JSONObject mediaJSON = data.getJSONObject(0);
JSONArray tracks = mediaJSON.getJSONArray("tracks");
if (authority) {
streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks, addr, calld, true);

16
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java
View File

@@ -169,7 +169,7 @@ public class PlatformServiceImpl implements IPlatformService {
dynamicTask.stop(registerTaskKey);
// 注销旧的
try {
if (parentPlatformOld.isStatus()) {
if (parentPlatformOld.isStatus() && parentPlatformCatchOld != null) {
logger.info("保存平台{}时发现旧平台在线，发送注销命令", parentPlatformOld.getServerGBId());
commanderForPlatform.unregister(parentPlatformOld, parentPlatformCatchOld.getSipTransactionInfo(), null, eventResult -> {
logger.info("[国标级联] 注销成功， 平台：{}", parentPlatformOld.getServerGBId());
@@ -286,6 +286,7 @@ public class PlatformServiceImpl implements IPlatformService {
}
if (parentPlatform.isAutoPushChannel()) {
if (subscribeHolder.getCatalogSubscribe(parentPlatform.getServerGBId()) == null) {
logger.info("[国标级联]：{}, 添加自动通道推送模拟订阅信息", parentPlatform.getServerGBId());
addSimulatedSubscribeInfo(parentPlatform);
}
}else {
@@ -363,9 +364,16 @@ public class PlatformServiceImpl implements IPlatformService {
// 清除心跳任务
dynamicTask.stop(keepaliveTaskKey);
}
// 停止目录订阅回复
logger.info("[平台离线] {}, 停止订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeAllSubscribe(parentPlatform.getServerGBId());
// 停止订阅回复
SubscribeInfo catalogSubscribe = subscribeHolder.getCatalogSubscribe(parentPlatform.getServerGBId());
if (catalogSubscribe != null) {
if (catalogSubscribe.getExpires() > 0) {
logger.info("[平台离线] {}, 停止目录订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeCatalogSubscribe(parentPlatform.getServerGBId());
}
}
logger.info("[平台离线] {}, 停止移动位置订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeMobilePositionSubscribe(parentPlatform.getServerGBId());
// 发起定时自动重新注册
if (!stopRegister) {
// 设置为60秒自动尝试重新注册

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

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.service.impl;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus;
@@ -19,13 +20,19 @@ import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
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.*;
import com.genersoft.iot.vmp.media.zlm.*;
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.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.*;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg;
@@ -33,6 +40,8 @@ import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
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.CloudRecordUtils;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
@@ -89,12 +98,18 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private IInviteStreamService inviteStreamService;
@Autowired
private ZlmHttpHookSubscribe subscribe;
@Autowired
private SendRtpPortManager sendRtpPortManager;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private ZLMServerFactory zlmServerFactory;
@Autowired
private AssistRESTfulUtils assistRESTfulUtils;
@@ -117,7 +132,7 @@ public class PlayServiceImpl implements IPlayService {
private DynamicTask dynamicTask;
@Autowired
private ZlmHttpHookSubscribe subscribe;
private CloudRecordServiceMapper cloudRecordServiceMapper;
@Autowired
private ISIPCommanderForPlatform commanderForPlatform;
@@ -407,6 +422,15 @@ public class PlayServiceImpl implements IPlayService {
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
}
}else {
logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},码流类型：{}，端口：{}, SSRC: {}",
device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流",
ssrcInfo.getPort(), ssrcInfo.getSsrc());
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
}
}, userSetting.getPlayTimeout());
@@ -437,6 +461,7 @@ public class PlayServiceImpl implements IPlayService {
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
timeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAY);
}, (event) -> {
logger.info("[点播失败] deviceId: {}, channelId:{}, {}: {}", device.getDeviceId(), channelId, event.statusCode, event.msg);
dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
@@ -478,7 +503,13 @@ public class PlayServiceImpl implements IPlayService {
if (!device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
return;
}
String substring = contentString.substring(0, contentString.indexOf("y="));
String substring;
if (contentString.indexOf("y=") > 0) {
substring = contentString.substring(0, contentString.indexOf("y="));
}else {
substring = contentString;
}
try {
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
int port = -1;
@@ -597,23 +628,6 @@ public class PlayServiceImpl implements IPlayService {
return mediaServerItem;
}
@Override
public MediaServerItem getNewMediaServerItemHasAssist(Device device) {
if (device == null) {
return null;
}
MediaServerItem mediaServerItem;
if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) {
mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(true);
} else {
mediaServerItem = mediaServerService.getOne(device.getMediaServerId());
}
if (mediaServerItem == null) {
logger.warn("[获取可用的ZLM节点]未找到可使用的ZLM...");
}
return mediaServerItem;
}
@Override
public void playBack(String deviceId, String channelId, String startTime,
String endTime, ErrorCallback<Object> callback) {
@@ -711,7 +725,6 @@ public class PlayServiceImpl implements IPlayService {
// 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
playBackTimeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAYBACK);
}, errorEvent);
} catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 录像回放: {}", e.getMessage());
@@ -732,6 +745,10 @@ public class PlayServiceImpl implements IPlayService {
ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
String contentString = new String(responseEvent.getResponse().getRawContent());
String ssrcInResponse = SipUtils.getSsrcFromSdp(contentString);
// 兼容回复的消息中缺少ssrc(y字段)的情况
if (ssrcInResponse == null) {
ssrcInResponse = ssrcInfo.getSsrc();
}
if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
// ssrc 一致
if (mediaServerItem.isRtpEnable()) {
@@ -809,13 +826,15 @@ public class PlayServiceImpl implements IPlayService {
}
@Override
public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, ErrorCallback<Object> callback) {
Device device = storager.queryVideoDevice(deviceId);
if (device == null) {
return;
}
MediaServerItem newMediaServerItem = getNewMediaServerItemHasAssist(device);
MediaServerItem newMediaServerItem = this.getNewMediaServerItem(device);
if (newMediaServerItem == null) {
callback.run(InviteErrorCode.ERROR_FOR_ASSIST_NOT_READY.getCode(),
InviteErrorCode.ERROR_FOR_ASSIST_NOT_READY.getMsg(),
@@ -894,6 +913,28 @@ public class PlayServiceImpl implements IPlayService {
// 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
downLoadTimeOutTaskKey, callback, inviteInfo, InviteSessionType.DOWNLOAD);
// 注册录像回调事件，录像下载结束后写入下载地址
ZlmHttpHookSubscribe.Event hookEventForRecord = (mediaServerItemInuse, hookParam) -> {
logger.info("[录像下载] 收到录像写入磁盘消息： ， {}/{}-{}",
inviteInfo.getDeviceId(), inviteInfo.getChannelId(), ssrcInfo.getStream());
logger.info("[录像下载] 收到录像写入磁盘消息内容： " + hookParam);
OnRecordMp4HookParam recordMp4HookParam = (OnRecordMp4HookParam)hookParam;
String filePath = recordMp4HookParam.getFile_path();
DownloadFileInfo downloadFileInfo = CloudRecordUtils.getDownloadFilePath(mediaServerItem, filePath);
InviteInfo inviteInfoForNew = inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId()
, inviteInfo.getChannelId(), inviteInfo.getStream());
inviteInfoForNew.getStreamInfo().setDownLoadFilePath(downloadFileInfo);
inviteStreamService.updateInviteInfo(inviteInfoForNew);
};
HookSubscribeForRecordMp4 hookSubscribe = HookSubscribeFactory.on_record_mp4(
mediaServerItem.getId(), "rtp", ssrcInfo.getStream());
// 设置过期时间，下载失败时自动处理订阅数据
// long difference = DateUtil.getDifference(startTime, endTime)/1000;
// Instant expiresInstant = Instant.now().plusSeconds(TimeUnit.MINUTES.toSeconds(difference * 2));
// hookSubscribe.setExpires(expiresInstant);
subscribe.addSubscribe(hookSubscribe, hookEventForRecord);
});
} catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 录像下载: {}", e.getMessage());
@@ -909,47 +950,71 @@ public class PlayServiceImpl implements IPlayService {
@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("[获取下载进度] 从zlm查询进度失败");
return null;
}
if (mediaListJson.getInteger("code") != 0) {
logger.warn("[获取下载进度] 从zlm查询进度出现错误： {}", mediaListJson.getString("msg"));
return null;
}
JSONArray data = mediaListJson.getJSONArray("data");
if (data == null) {
logger.warn("[获取下载进度] 从zlm查询进度时未返回数据");
return null;
}
JSONObject mediaJSON = data.getJSONObject(0);
JSONArray tracks = mediaJSON.getJSONArray("tracks");
if (tracks.isEmpty()) {
logger.warn("[获取下载进度] 从zlm查询进度时未返回数据");
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和end单位是秒
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();
if (process > 0.999) {
process = 1.0;
}
inviteInfo.getStreamInfo().setProgress(process);
}
inviteStreamService.updateInviteInfo(inviteInfo);
return inviteInfo.getStreamInfo();
}
private StreamInfo onPublishHandlerForDownload(MediaServerItem mediaServerItemInuse, HookParam hookParam, String deviceId, String channelId, String startTime, String endTime) {
@@ -1219,7 +1284,12 @@ public class PlayServiceImpl implements IPlayService {
throw new ServiceException("mediaServer不存在");
}
// zlm 暂停RTP超时检查
JSONObject jsonObject = zlmresTfulUtils.pauseRtpCheck(mediaServerItem, streamId);
// 使用zlm中的流ID
String streamKey = inviteInfo.getStream();
if (!mediaServerItem.isRtpEnable()) {
streamKey = Long.toHexString(Long.parseLong(inviteInfo.getSsrcInfo().getSsrc())).toUpperCase();
}
JSONObject jsonObject = zlmresTfulUtils.pauseRtpCheck(mediaServerItem, streamKey);
if (jsonObject == null || jsonObject.getInteger("code") != 0) {
throw new ServiceException("暂停RTP接收失败");
}
@@ -1242,7 +1312,12 @@ public class PlayServiceImpl implements IPlayService {
throw new ServiceException("mediaServer不存在");
}
// zlm 暂停RTP超时检查
JSONObject jsonObject = zlmresTfulUtils.resumeRtpCheck(mediaServerItem, streamId);
// 使用zlm中的流ID
String streamKey = inviteInfo.getStream();
if (!mediaServerItem.isRtpEnable()) {
streamKey = Long.toHexString(Long.parseLong(inviteInfo.getSsrcInfo().getSsrc())).toUpperCase();
}
JSONObject jsonObject = zlmresTfulUtils.resumeRtpCheck(mediaServerItem, streamKey);
if (jsonObject == null || jsonObject.getInteger("code") != 0) {
throw new ServiceException("继续RTP接收失败");
}

10
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
View File

@@ -126,7 +126,13 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}
JSONArray dataArray = jsonObject.getJSONArray("data");
JSONObject mediaServerConfig = dataArray.getJSONObject(0);
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拉流代理无法获取ffmpeg cmd");
}
String schema = getSchemaFromFFmpegCmd(ffmpegCmd);
if (schema == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "ffmpeg拉流代理无法从ffmpeg cmd中获取到输出格式");
@@ -401,6 +407,8 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
logger.info("启用代理失败： {}/{}->{}({})", app, stream, jsonObject.getString("msg"),
streamProxy.getSrcUrl() == null? streamProxy.getUrl():streamProxy.getSrcUrl());
}
} else if (streamProxy != null && streamProxy.isEnable()) {
return true ;
}
return result;
}
@@ -452,7 +460,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
streamProxyMapper.deleteAutoRemoveItemByMediaServerId(mediaServerId);
// 移除拉流代理生成的流信息
// syncPullStream(mediaServerId);
syncPullStream(mediaServerId);
// 恢复流代理, 只查找这个这个流媒体
List<StreamProxyItem> streamProxyListForEnable = storager.getStreamProxyListForEnableInMediaServer(

5
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
View File

@@ -282,6 +282,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream());
// 冗余数据，自己系统中自用
redisCatchStorage.removePushListItem(offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream(), mediaServerItem.getId());
}
}
@@ -319,6 +321,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 冗余数据，自己系统中自用
redisCatchStorage.removePushListItem(onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream(), mediaServerId);
}
}
}

18
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
View File

@@ -113,8 +113,8 @@ public class RedisGbPlayMsgListener implements MessageListener {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
try {
JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
WvpRedisMsg wvpRedisMsg = JSON.parseObject(msg.getBody(), WvpRedisMsg.class);
logger.info("[收到REDIS通知] 消息： {}", JSON.toJSONString(wvpRedisMsg));
if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
continue;
}
@@ -123,7 +123,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
switch (wvpRedisMsg.getCmd()){
case WvpRedisMsgCmd.GET_SEND_ITEM:
RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
RequestSendItemMsg content = JSON.parseObject(wvpRedisMsg.getContent(), RequestSendItemMsg.class);
requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break;
case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
@@ -242,7 +242,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
result.setData(content);
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, JSON.toJSONString(result));
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
@@ -260,7 +260,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
result.setMsg("流媒体不存在");
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);
WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result));
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -283,7 +283,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
WVPResult<SendRtpItem> result = new WVPResult<>();
result.setCode(ERROR_CODE_TIMEOUT);
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -324,7 +324,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
result.setData(responseSendItemMsg);
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -350,7 +350,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
requestSendItemMsg.setServerId(serverId);
String key = UUID.randomUUID().toString();
WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId, WvpRedisMsgCmd.GET_SEND_ITEM,
key, requestSendItemMsg);
key, JSON.toJSONString(requestSendItemMsg));
JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
logger.info("[请求推流SendItem] {}: {}", serverId, jsonObject);
@@ -375,7 +375,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
public void sendMsgForStartSendRtpStream(String serverId, RequestPushStreamMsg param, PlayMsgCallbackForStartSendRtpStream callback) {
String key = UUID.randomUUID().toString();
WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, param);
WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, JSON.toJSONString(param));
JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
logger.info("[REDIS 请求其他平台推流] {}: {}", serverId, jsonObject);

5
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java
View File

@@ -50,11 +50,12 @@ public class RedisGpsMsgListener implements MessageListener {
Message msg = taskQueue.poll();
try {
GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class);
logger.info("[REDIS的位置变化通知], {}", JSON.toJSONString(gpsMsgInfo));
// 只是放入redis缓存起来
redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo);
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
logger.error("[REDIS的ALARM通知] 异常内容： ", e);
logger.warn("[REDIS的位置变化通知] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
logger.error("[REDIS的位置变化通知] 异常内容： ", e);
}
}
});