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

1078-添加录像回放开始和停止

Browse Source
This commit is contained in:
648540858
2024-04-06 00:06:40 +08:00
parent 81c2b5715c
commit 26bbeac6c7
8 changed files with 306 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
View File

@@ -178,6 +178,10 @@ public class VideoManagerConstants {
//************************** 1078 ****************************************
public static final String INVITE_INFO_1078 = "INVITE_INFO_1078:";
public static final String INVITE_INFO_1078_PLAY = "INVITE_INFO_1078_PLAY:";
public static final String INVITE_INFO_1078_PLAYBACK = "INVITE_INFO_1078_PLAYBACK:";
public static final String RECORD_LIST_1078 = "RECORD_LIST_1078:";
}

92
src/main/java/com/genersoft/iot/vmp/jt1078/config/JT1078Controller.java
View File

@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.jt1078.config;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.jt1078.bean.JTDevice;
import com.genersoft.iot.vmp.jt1078.proc.request.J1205;
import com.genersoft.iot.vmp.jt1078.service.Ijt1078Service;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
@@ -24,6 +25,7 @@ import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.List;
/**
* curl http://localhost:18080/api/jt1078/start/live/18864197066/1
@@ -139,22 +141,98 @@ public class JT1078Controller {
service.continueLivePlay(deviceId, channelId);
}
@Operation(summary = "1078-回放-查询资源列表", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Operation(summary = "1078-录像-查询资源列表", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "channelId", description = "通道国标编号, 一般为从1开始的数字", required = true)
@Parameter(name = "startTime", description = "开始时间,格式： yyyy-MM-dd HH:mm:ss", required = true)
@Parameter(name = "endTime", description = "结束时间,格式： yyyy-MM-dd HH:mm:ss", required = true)
@GetMapping("/record/list")
public void playbackList(HttpServletRequest request,
@Parameter(required = true) String deviceId,
@Parameter(required = false) String channelId,
@Parameter(required = true) String startTime,
@Parameter(required = true) String endTime
public WVPResult<List<J1205.JRecordItem>> playbackList(HttpServletRequest request,
@Parameter(required = true) String deviceId,
@Parameter(required = false) String channelId,
@Parameter(required = true) String startTime,
@Parameter(required = true) String endTime
) {
if (ObjectUtils.isEmpty(channelId)) {
channelId = "1";
}
service.getRecordList(deviceId, channelId, startTime, endTime);
List<J1205.JRecordItem> recordList = service.getRecordList(deviceId, channelId, startTime, endTime);
if (recordList == null) {
return WVPResult.fail(ErrorCode.ERROR100);
}else {
return WVPResult.success(recordList);
}
}
@Operation(summary = "1078-开始回放", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "channelId", description = "通道国标编号, 一般为从1开始的数字", required = true)
@Parameter(name = "startTime", description = "开始时间,格式： yyyy-MM-dd HH:mm:ss", required = true)
@Parameter(name = "endTime", description = "结束时间,格式： yyyy-MM-dd HH:mm:ss", required = true)
@GetMapping("/playback/start")
public DeferredResult<WVPResult<StreamContent>> recordLive(HttpServletRequest request,
@Parameter(required = true) String deviceId,
@Parameter(required = false) String channelId,
@Parameter(required = true) String startTime,
@Parameter(required = true) String endTime) {
DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
if (ObjectUtils.isEmpty(channelId)) {
channelId = "1";
}
String finalChannelId = channelId;
result.onTimeout(()->{
logger.info("[1078-回放-等待超时] deviceId：{}, channelId：{}, ", deviceId, finalChannelId);
// 释放rtpserver
WVPResult<StreamContent> wvpResult = new WVPResult<>();
wvpResult.setCode(ErrorCode.ERROR100.getCode());
wvpResult.setMsg("回放超时");
result.setResult(wvpResult);
service.stopPlay(deviceId, finalChannelId);
});
service.playback(deviceId, channelId, startTime, endTime, (code, msg, streamInfo) -> {
WVPResult<StreamContent> wvpResult = new WVPResult<>();
if (code == InviteErrorCode.SUCCESS.getCode()) {
wvpResult.setCode(ErrorCode.SUCCESS.getCode());
wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
if (streamInfo != null) {
if (userSetting.getUseSourceIpAsStreamIp()) {
streamInfo=streamInfo.clone();//深拷贝
String host;
try {
URL url=new URL(request.getRequestURL().toString());
host=url.getHost();
} catch (MalformedURLException e) {
host=request.getLocalAddr();
}
streamInfo.channgeStreamIp(host);
}
wvpResult.setData(new StreamContent(streamInfo));
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
}
}else {
wvpResult.setCode(code);
wvpResult.setMsg(msg);
}
result.setResult(wvpResult);
});
return result;
}
@Operation(summary = "1078-结束回放", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "channelId", description = "通道国标编号, 一般为从1开始的数字", required = true)
@GetMapping("/playback/stop")
public void stopPlayback(HttpServletRequest request,
@Parameter(required = true) String deviceId,
@Parameter(required = false) String channelId) {
if (ObjectUtils.isEmpty(channelId)) {
channelId = "1";
}
service.stopPlayback(deviceId, channelId);
}
@Operation(summary = "分页查询部标设备", security = @SecurityRequirement(name = JwtUtils.HEADER))

69
src/main/java/com/genersoft/iot/vmp/jt1078/event/CallbackManager.java Normal file
View File

@@ -0,0 +1,69 @@
package com.genersoft.iot.vmp.jt1078.event;
import com.genersoft.iot.vmp.common.GeneralCallback;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 管理callback回调，支持设置超时时间，未设置则按照五分钟超时自动移除
*/
@Component
public class CallbackManager {
private final long expire = 5 * 60 * 1000;
static class ManagerCallBack {
public String key;
public GeneralCallback<?> callback;
public long createTime;
public long expire;
}
private final Map<String, ManagerCallBack> allCallbacks = new ConcurrentHashMap<>();
public void addCallback(String key, GeneralCallback<?> callback) {
ManagerCallBack managerCallBack = new ManagerCallBack();
managerCallBack.callback = callback;
managerCallBack.key = key;
managerCallBack.createTime = System.currentTimeMillis();
managerCallBack.expire = expire;
allCallbacks.put(key, managerCallBack);
}
public void addCallback(String key, GeneralCallback<Object> callback, long timeout) {
ManagerCallBack managerCallBack = new ManagerCallBack();
managerCallBack.callback = callback;
managerCallBack.key = key;
managerCallBack.createTime = System.currentTimeMillis();
managerCallBack.expire = timeout;
allCallbacks.put(key, managerCallBack);
}
public GeneralCallback<?> getCallback(String key){
ManagerCallBack managerCallBack = allCallbacks.get(key);
if (managerCallBack != null) {
return managerCallBack.callback;
}else {
return null;
}
}
public void removeCallback(String key){
allCallbacks.remove(key);
}
/**
* 对订阅数据进行过期清理
*/
@Scheduled(fixedRate=expire) //每5分钟执行一次
public void execute(){
for (ManagerCallBack callBack : allCallbacks.values()) {
if ((System.currentTimeMillis() - callBack.createTime - callBack.expire) > 0) {
allCallbacks.remove(callBack.key);
}
}
}
}

2
src/main/java/com/genersoft/iot/vmp/jt1078/proc/request/J1205.java
View File

@@ -45,14 +45,12 @@ public class J1205 extends Re {
item.setSize(buf.readUnsignedInt());
recordList.add(item);
}
return null;
}
@Override
protected Rs handler(Header header, Session session, Ijt1078Service service) {
SessionManager.INSTANCE.response(header.getTerminalId(), "1205", (long) respNo, JSON.toJSONString(this));
service.notifyRecordList()
J8001 j8001 = new J8001();
j8001.setRespNo(header.getSn());
j8001.setRespId(header.getMsgId());

9
src/main/java/com/genersoft/iot/vmp/jt1078/service/Ijt1078Service.java
View File

@@ -3,8 +3,11 @@ package com.genersoft.iot.vmp.jt1078.service;
import com.genersoft.iot.vmp.common.GeneralCallback;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.jt1078.bean.JTDevice;
import com.genersoft.iot.vmp.jt1078.proc.request.J1205;
import com.github.pagehelper.PageInfo;
import java.util.List;
public interface Ijt1078Service {
JTDevice getDevice(String terminalId);
@@ -20,11 +23,15 @@ public interface Ijt1078Service {
void play(String deviceId, String channelId, GeneralCallback<StreamInfo> callback);
void playback(String deviceId, String channelId, String startTime, String endTime, GeneralCallback<StreamInfo> callback);
void stopPlay(String deviceId, String channelId);
void pausePlay(String deviceId, String channelId);
void continueLivePlay(String deviceId, String channelId);
void getRecordList(String deviceId, String channelId, String startTime, String endTime);
List<J1205.JRecordItem> getRecordList(String deviceId, String channelId, String startTime, String endTime);
void stopPlayback(String deviceId, String channelId);
}

151
src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java
View File

@@ -1,18 +1,18 @@
package com.genersoft.iot.vmp.jt1078.service.impl;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.GeneralCallback;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.jt1078.bean.JTDevice;
import com.genersoft.iot.vmp.jt1078.cmd.JT1078Template;
import com.genersoft.iot.vmp.jt1078.config.JT1078Controller;
import com.genersoft.iot.vmp.jt1078.dao.JTDeviceMapper;
import com.genersoft.iot.vmp.jt1078.proc.response.J9101;
import com.genersoft.iot.vmp.jt1078.proc.response.J9102;
import com.genersoft.iot.vmp.jt1078.event.CallbackManager;
import com.genersoft.iot.vmp.jt1078.proc.request.J1205;
import com.genersoft.iot.vmp.jt1078.proc.response.*;
import com.genersoft.iot.vmp.jt1078.service.Ijt1078Service;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
@@ -23,24 +23,17 @@ 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.IMediaServerService;
import com.genersoft.iot.vmp.service.IMediaService;
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.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
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.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -78,6 +71,9 @@ public class jt1078ServiceImpl implements Ijt1078Service {
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private CallbackManager callbackManager;
@Override
public JTDevice getDevice(String terminalId) {
@@ -120,7 +116,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
public void play(String deviceId, String channelId, GeneralCallback<StreamInfo> callback) {
// 检查流是否已经存在，存在则返回
String playKey = VideoManagerConstants.INVITE_INFO_1078 + deviceId + ":" + channelId;
String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAY + deviceId + ":" + channelId;
List<GeneralCallback<StreamInfo>> errorCallbacks = inviteErrorCallbackMap.computeIfAbsent(playKey, k -> new ArrayList<>());
errorCallbacks.add(callback);
StreamInfo streamInfo = (StreamInfo)redisTemplate.opsForValue().get(playKey);
@@ -195,7 +191,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
@Override
public void stopPlay(String deviceId, String channelId) {
String playKey = VideoManagerConstants.INVITE_INFO_1078 + deviceId + ":" + channelId;
String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAY + deviceId + ":" + channelId;
dynamicTask.stop(playKey);
StreamInfo streamInfo = (StreamInfo) redisTemplate.opsForValue().get(playKey);
// 发送停止命令
@@ -222,7 +218,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
@Override
public void pausePlay(String deviceId, String channelId) {
String playKey = VideoManagerConstants.INVITE_INFO_1078 + deviceId + ":" + channelId;
String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAY + deviceId + ":" + channelId;
dynamicTask.stop(playKey);
StreamInfo streamInfo = (StreamInfo) redisTemplate.opsForValue().get(playKey);
if (streamInfo == null) {
@@ -240,7 +236,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
@Override
public void continueLivePlay(String deviceId, String channelId) {
String playKey = VideoManagerConstants.INVITE_INFO_1078 + deviceId + ":" + channelId;
String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAY + deviceId + ":" + channelId;
dynamicTask.stop(playKey);
StreamInfo streamInfo = (StreamInfo) redisTemplate.opsForValue().get(playKey);
if (streamInfo == null) {
@@ -257,7 +253,130 @@ public class jt1078ServiceImpl implements Ijt1078Service {
}
@Override
public void getRecordList(String deviceId, String channelId, String startTime, String endTime) {
public List<J1205.JRecordItem> getRecordList(String deviceId, String channelId, String startTime, String endTime) {
logger.info("[1078-查询录像列表] deviceId： {}， channelId： {}， startTime： {}， endTime： {}"
, deviceId, channelId, startTime, endTime);
// 发送请求录像列表命令
J9205 j9205 = new J9205();
j9205.setChannelId(Integer.parseInt(channelId));
j9205.setStartTime(DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(startTime));
j9205.setEndTime(DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(endTime));
j9205.setMediaType(0);
j9205.setStreamType(0);
j9205.setStorageType(0);
String J1205JSON = jt1078Template.queryBackTime(deviceId, j9205, 20);
if (J1205JSON == null) {
return null;
}
J1205 j1205 = JSON.parseObject(J1205JSON, J1205.class);
if (j1205 == null) {
return null;
}
logger.info("[1078-查询录像列表] deviceId： {}， channelId： {}， startTime： {}， endTime： {}, 结果: {}条"
, deviceId, channelId, startTime, endTime, j1205.getRecordList().size() );
return j1205.getRecordList();
}
@Override
public void playback(String deviceId, String channelId, String startTime, String endTime, GeneralCallback<StreamInfo> callback) {
// 检查流是否已经存在，存在则返回
String playbackKey = VideoManagerConstants.INVITE_INFO_1078_PLAYBACK + deviceId + ":" + channelId;
List<GeneralCallback<StreamInfo>> errorCallbacks = inviteErrorCallbackMap.computeIfAbsent(playbackKey, k -> new ArrayList<>());
errorCallbacks.add(callback);
String logInfo = String.format("deviceId:%s, channelId:%s, startTime:%s, endTime:%s", deviceId, channelId, startTime, endTime);
StreamInfo streamInfo = (StreamInfo)redisTemplate.opsForValue().get(playbackKey);
if (streamInfo != null) {
String mediaServerId = streamInfo.getMediaServerId();
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem != null) {
// 查询流是否存在，不存在则删除缓存数据
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, "rtp", "rtsp", streamInfo.getStream());
if (mediaInfo != null && mediaInfo.getInteger("code") == 0 ) {
Boolean online = mediaInfo.getBoolean("online");
if (online != null && online) {
logger.info("[1078-回放] 回放已经存在，直接返回， logInfo： {}", logInfo);
for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {
errorCallback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
}
return;
}
}
}
// 清理数据
redisTemplate.delete(playbackKey);
}
String stream = deviceId + "-" + channelId + "-" + startTime + "-" + endTime;
MediaServerItem mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null);
// 设置hook监听
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> {
OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam) hookParam;
dynamicTask.stop(playbackKey);
logger.info("[1078-回放] 回放成功， logInfo： {}", logInfo);
StreamInfo info = onPublishHandler(mediaServerItemInUse, streamChangedHookParam, deviceId, channelId);
for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {
errorCallback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), info);
}
subscribe.removeSubscribe(hookSubscribe);
redisTemplate.opsForValue().set(playbackKey, info);
});
// 设置超时监听
dynamicTask.startDelay(playbackKey, () -> {
logger.info("[1078-回放] 回放超时， logInfo： {}", logInfo);
for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {
errorCallback.run(InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode(),
InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getMsg(), null);
}
}, userSetting.getPlayTimeout());
// 开启收流端口
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, false, false, 0, false, false, 1);
logger.info("[1078-回放] logInfo： {}， 端口： {}", logInfo, ssrcInfo.getPort());
J9201 j9201 = new J9201();
j9201.setChannel(Integer.parseInt(channelId));
j9201.setIp(mediaServerItem.getSdpIp());
j9201.setRate(1);
j9201.setPlaybackType(0);
j9201.setPlaybackSpeed(0);
j9201.setTcpPort(ssrcInfo.getPort());
j9201.setUdpPort(ssrcInfo.getPort());
j9201.setType(0);
j9201.setStartTime(DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(startTime));
j9201.setEndTime(DateUtil.yyyy_MM_dd_HH_mm_ssToUrl(endTime));
String s = jt1078Template.startBackLive(deviceId, j9201, 6);
System.out.println("111ssss=== " + s);
}
@Override
public void stopPlayback(String deviceId, String channelId) {
String playKey = VideoManagerConstants.INVITE_INFO_1078_PLAYBACK + deviceId + ":" + channelId;
dynamicTask.stop(playKey);
StreamInfo streamInfo = (StreamInfo) redisTemplate.opsForValue().get(playKey);
// 发送停止命令
J9202 j9202 = new J9202();
j9202.setChannel(Integer.parseInt(channelId));
j9202.setPlaybackType(0);
j9202.setPlaybackSpeed(0);
j9202.setPlaybackTime("");
jt1078Template.controlBackLive(deviceId, j9202, 6);
logger.info("[1078-停止回放] deviceId： {}， channelId： {}", deviceId, channelId);
// 删除缓存数据
if (streamInfo != null) {
// 关闭rtpServer
mediaServerService.closeRTPServer(streamInfo.getMediaServerId(), streamInfo.getStream());
}
// 清理回调
List<GeneralCallback<StreamInfo>> generalCallbacks = inviteErrorCallbackMap.get(playKey);
if (!generalCallbacks.isEmpty()) {
for (GeneralCallback<StreamInfo> callback : generalCallbacks) {
callback.run(InviteErrorCode.ERROR_FOR_FINISH.getCode(), InviteErrorCode.ERROR_FOR_FINISH.getMsg(), null);
}
}
}
}

3
src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java
View File

@@ -86,6 +86,9 @@ public class DateUtil {
public static String urlToyyyy_MM_dd_HH_mm_ss(String formatTime) {
return formatter.format(urlFormatter.parse(formatTime));
}
public static String yyyy_MM_dd_HH_mm_ssToUrl(String formatTime) {
return urlFormatter.format(formatter.parse(formatTime));
}
/**
* yyyy_MM_dd_HH_mm_ss 转时间戳

1
src/main/java/com/genersoft/iot/vmp/vmanager/bean/ErrorCode.java
View File

@@ -10,6 +10,7 @@ public enum ErrorCode {
ERROR404(404, "资源未找到"),
ERROR403(403, "无权限操作"),
ERROR401(401, "请登录后重新请求"),
ERROR408(408, "请求超时"),
ERROR500(500, "系统异常");
private final int code;