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 wvp-jwt-token

Browse Source 
# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
#	src/main/resources/all-application.yml
This commit is contained in:
648540858
2023-03-15 22:58:28 +08:00
parent efbc3a31c0 be6666db03
commit 2bf07305aa
30 changed files with 320 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
View File

@@ -48,6 +48,7 @@ public class UserSetting {
private Boolean syncChannelOnDeviceOnline = Boolean.FALSE;
private Boolean sipLog = Boolean.FALSE;
private Boolean sendToPlatformsWhenIdLost = Boolean.FALSE;
private String serverId = "000000";
@@ -228,4 +229,12 @@ public class UserSetting {
public void setAllowedOrigins(List<String> allowedOrigins) {
this.allowedOrigins = allowedOrigins;
}
public Boolean getSendToPlatformsWhenIdLost() {
return sendToPlatformsWhenIdLost;
}
public void setSendToPlatformsWhenIdLost(Boolean sendToPlatformsWhenIdLost) {
this.sendToPlatformsWhenIdLost = sendToPlatformsWhenIdLost;
}
}

11
src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java
View File

@@ -188,6 +188,9 @@ public class Device {
@Schema(description = "SIP交互IP（设备访问平台的IP）")
private String localIp;
@Schema(description = "是否作为消息通道")
private boolean asMessageChannel;
public String getDeviceId() {
return deviceId;
@@ -428,4 +431,12 @@ public class Device {
public void setKeepaliveIntervalTime(int keepaliveIntervalTime) {
this.keepaliveIntervalTime = keepaliveIntervalTime;
}
public boolean isAsMessageChannel() {
return asMessageChannel;
}
public void setAsMessageChannel(boolean asMessageChannel) {
this.asMessageChannel = asMessageChannel;
}
}

11
src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java
View File

@@ -189,6 +189,9 @@ public class ParentPlatform {
@Schema(description = "树类型 国标规定了两种树的展现方式 行政区划 CivilCode 和业务分组:BusinessGrou")
private String treeType;
@Schema(description = "是否作为消息通道")
private boolean asMessageChannel;
public Integer getId() {
return id;
}
@@ -428,4 +431,12 @@ public class ParentPlatform {
public void setTreeType(String treeType) {
this.treeType = treeType;
}
public boolean isAsMessageChannel() {
return asMessageChannel;
}
public void setAsMessageChannel(boolean asMessageChannel) {
this.asMessageChannel = asMessageChannel;
}
}

13
src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java
View File

@@ -1,6 +1,8 @@
package com.genersoft.iot.vmp.gb28181.bean;
import io.swagger.v3.oas.annotations.media.Schema;
import java.time.Instant;
import java.util.List;
@@ -9,22 +11,29 @@ import java.util.List;
* @author: swwheihei
* @date: 2020年5月8日 下午2:05:56
*/
@Schema(description = "设备录像查询结果信息")
public class RecordInfo {
@Schema(description = "设备编号")
private String deviceId;
@Schema(description = "通道编号")
private String channelId;
@Schema(description = "命令序列号")
private String sn;
@Schema(description = "设备名称")
private String name;
@Schema(description = "列表总数")
private int sumNum;
private int count;
private Instant lastTime;
@Schema(description = "")
private List<RecordItem> recordList;
public String getDeviceId() {

27
src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordItem.java
View File

@@ -2,9 +2,9 @@ package com.genersoft.iot.vmp.gb28181.bean;
import com.genersoft.iot.vmp.utils.DateUtil;
import io.swagger.v3.oas.annotations.media.Schema;
import org.jetbrains.annotations.NotNull;
import java.text.ParseException;
import java.time.Instant;
import java.time.temporal.TemporalAccessor;
@@ -13,26 +13,37 @@ import java.time.temporal.TemporalAccessor;
* @author: swwheihei
* @date: 2020年5月8日 下午2:06:54
*/
@Schema(description = "设备录像详情")
public class RecordItem implements Comparable<RecordItem>{
@Schema(description = "设备编号")
private String deviceId;
@Schema(description = "名称")
private String name;
@Schema(description = "文件路径名 (可选)")
private String filePath;
@Schema(description = "录像文件大小,单位:Byte(可选)")
private String fileSize;
@Schema(description = "录像地址(可选)")
private String address;
@Schema(description = "录像开始时间(可选)")
private String startTime;
@Schema(description = "录像结束时间(可选)")
private String endTime;
@Schema(description = "保密属性(必选)缺省为0;0:不涉密,1:涉密")
private int secrecy;
@Schema(description = "录像产生类型(可选)time或alarm 或 manua")
private String type;
@Schema(description = "录像触发者ID(可选)")
private String recorderId;
public String getDeviceId() {

3
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
View File

@@ -122,7 +122,7 @@ public interface ISIPCommander {
*/
void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, int downloadSpeed, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException;
SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException;
/**
* 视频流停止
@@ -221,7 +221,6 @@ public interface ISIPCommander {
*
* @param device 视频设备
* @param channelId 通道id，非通道则是设备本身
* @param frontCmd 上级平台的指令，如果存在则直接下发
* @param enabled 看守位使能：1 = 开启，0 = 关闭
* @param resetTime 自动归位时间间隔，开启看守位时使用，单位:秒(s)
* @param presetIndex 调用预置位编号，开启看守位时使用，取值范围0~255

36
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@@ -470,8 +470,9 @@ public class SIPCommander implements ISIPCommander {
*/
@Override
public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
String startTime, String endTime, int downloadSpeed, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException {
String startTime, String endTime, int downloadSpeed,
InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException {
logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
String sdpIp;
@@ -540,11 +541,14 @@ public class SIPCommander implements ISIPCommander {
content.append("a=downloadspeed:" + downloadSpeed + "\r\n");
content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc
logger.debug("此时请求下载信令的ssrc===>{}",ssrcInfo.getSsrc());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, null, mediaServerItem.getId());
// 添加订阅
CallIdHeader newCallIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());
String callId=newCallIdHeader.getCallId();
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
hookEvent.call(new InviteStreamInfo(mediaServerItem, json,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
logger.debug("sipc 添加订阅===callId {}",callId);
hookEvent.call(new InviteStreamInfo(mediaServerItem, json,callId, "rtp", ssrcInfo.getStream()));
subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("regist", false);
hookSubscribe.getContent().put("schema", "rtsp");
@@ -553,7 +557,7 @@ public class SIPCommander implements ISIPCommander {
(MediaServerItem mediaServerItemForEnd, JSONObject jsonForEnd) -> {
logger.info("[录像]下载结束， 发送BYE");
try {
streamByeCmd(device, channelId, ssrcInfo.getStream(),sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId());
streamByeCmd(device, channelId, ssrcInfo.getStream(),callId);
} catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) {
logger.error("[录像]下载结束， 发送BYE失败 {}", e.getMessage());
@@ -561,15 +565,24 @@ public class SIPCommander implements ISIPCommander {
});
});
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()), ssrcInfo.getSsrc());
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,newCallIdHeader, ssrcInfo.getSsrc());
if (inviteStreamCallback != null) {
inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,callId, "rtp", ssrcInfo.getStream()));
}
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, okEvent -> {
ResponseEvent responseEvent = (ResponseEvent) okEvent.event;
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
ResponseEvent responseEvent = (ResponseEvent) event.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse();
streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
String contentString =new String(response.getRawContent());
int ssrcIndex = contentString.indexOf("y=");
String ssrc=ssrcInfo.getSsrc();
if (ssrcIndex >= 0) {
ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
}
logger.debug("接收到的下载响应ssrc====>{}",ssrcInfo.getSsrc());
logger.debug("接收到的下载响应ssrc====>{}",ssrc);
streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
okEvent.response(event);
});
}
@@ -801,7 +814,6 @@ public class SIPCommander implements ISIPCommander {
*
* @param device 视频设备
* @param channelId 通道id，非通道则是设备本身
* @param frontCmd 上级平台的指令，如果存在则直接下发
* @param enabled 看守位使能：1 = 开启，0 = 关闭
* @param resetTime 自动归位时间间隔，开启看守位时使用，单位:秒(s)
* @param presetIndex 调用预置位编号，开启看守位时使用，取值范围0~255
@@ -1376,7 +1388,7 @@ public class SIPCommander implements ISIPCommander {
if (device == null) {
return;
}
logger.info("[发送 报警通知] {}/{}->{},{}", device.getDeviceId(), deviceAlarm.getChannelId(),
logger.info("[发送报警通知]设备： {}/{}->{},{}", device.getDeviceId(), deviceAlarm.getChannelId(),
deviceAlarm.getLongitude(), deviceAlarm.getLatitude());
String characterSet = device.getCharset();

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
View File

@@ -402,7 +402,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if (parentPlatform == null) {
return;
}
logger.info("[发送报警通知] {}/{}->{},{}: {}", parentPlatform.getServerGBId(), deviceAlarm.getChannelId(),
logger.info("[发送报警通知]平台： {}/{}->{},{}: {}", parentPlatform.getServerGBId(), deviceAlarm.getChannelId(),
deviceAlarm.getLongitude(), deviceAlarm.getLatitude(), JSON.toJSONString(deviceAlarm));
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);

6
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
View File

@@ -163,7 +163,11 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
public void releaseRequest(String deviceId, String sn,RecordInfo recordInfo){
String key = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + deviceId + sn;
// 对数据进行排序
Collections.sort(recordInfo.getRecordList());
if(recordInfo!=null && recordInfo.getRecordList()!=null) {
Collections.sort(recordInfo.getRecordList());
}else{
recordInfo.setRecordList(new ArrayList<>());
}
RequestMessage msg = new RequestMessage();
msg.setKey(key);

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

@@ -256,6 +256,7 @@ public class ZLMHttpHookListener {
return result;
}
/**
* rtsp/rtmp流注册或注销时触发此事件；此事件对回复不敏感。
*/
@@ -279,9 +280,12 @@ public class ZLMHttpHookListener {
subscribe.response(mediaInfo, json);
}
}
// 流消失移除redis play
List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
// TODO 重构此处逻辑
if (param.isRegist()) {
// 处理流注册的鉴权信息
if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
|| param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
@@ -300,16 +304,15 @@ public class ZLMHttpHookListener {
}
if ("rtsp".equals(param.getSchema())) {
// 更新流媒体负载信息
if (param.isRegist()) {
mediaServerService.addCount(param.getMediaServerId());
} else {
mediaServerService.removeCount(param.getMediaServerId());
}
if (param.getOriginType() == OriginType.PULL.ordinal()
|| param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
// 设置拉流代理上线/离线
streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
}
// 设置拉流代理上线/离线
streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
if ("rtp".equals(param.getApp()) && !param.isRegist()) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());
if (streamInfo != null) {

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

@@ -635,23 +635,75 @@ public class PlayServiceImpl implements IPlayService {
hookCallBack.call(downloadResult);
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
};
InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getCallId());
dynamicTask.stop(downLoadTimeOutTaskKey);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
streamInfo.setStartTime(startTime);
streamInfo.setEndTime(endTime);
redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
downloadResult.setCode(ErrorCode.SUCCESS.getCode());
downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
downloadResult.setData(streamInfo);
downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
downloadResult.setResponse(inviteStreamInfo.getResponse());
hookCallBack.call(downloadResult);
};
try {
cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack,
inviteStreamInfo -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
dynamicTask.stop(downLoadTimeOutTaskKey);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
streamInfo.setStartTime(startTime);
streamInfo.setEndTime(endTime);
redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
downloadResult.setCode(ErrorCode.SUCCESS.getCode());
downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
downloadResult.setData(streamInfo);
downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
downloadResult.setResponse(inviteStreamInfo.getResponse());
hookCallBack.call(downloadResult);
}, errorEvent);
hookEvent, errorEvent, eventResult ->
{
if (eventResult.type == SipSubscribe.EventResultType.response) {
ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
String contentString = new String(responseEvent.getResponse().getRawContent());
// 获取ssrc
int ssrcIndex = contentString.indexOf("y=");
// 检查是否有y字段
if (ssrcIndex >= 0) {
//ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
// 查询到ssrc不一致且开启了ssrc校验则需要针对处理
if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
return;
}
logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
// ssrc 不可用
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
eventResult.msg = "下级自定义了ssrc,但是此ssrc不可用";
eventResult.statusCode = 400;
errorEvent.response(eventResult);
return;
}
// 单端口模式streamId也有变化，需要重新设置监听
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString());
dynamicTask.stop(downLoadTimeOutTaskKey);
// hook响应
onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, hookCallBack);
hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
});
}
// 关闭rtp server
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 重新开启ssrc server
mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort());
}
}
}
});
} catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 录像下载: {}", e.getMessage());

46
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisAlarmMsgListener.java
View File

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
@@ -44,8 +45,12 @@ public class RedisAlarmMsgListener implements MessageListener {
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
@Autowired
private UserSetting userSetting;
@Override
public void onMessage(@NotNull Message message, byte[] bytes) {
// 消息示例： PUBLISH alarm_receive '{ "gbId": "", "alarmSn": 1, "alarmType": "111", "alarmDescription": "222", }'
logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody()));
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message);
@@ -74,17 +79,44 @@ public class RedisAlarmMsgListener implements MessageListener {
deviceAlarm.setLatitude(0);
if (ObjectUtils.isEmpty(gbId)) {
// 发送给所有的上级
List<ParentPlatform> parentPlatforms = storage.queryEnableParentPlatformList(true);
if (parentPlatforms.size() > 0) {
for (ParentPlatform parentPlatform : parentPlatforms) {
if (userSetting.getSendToPlatformsWhenIdLost()) {
// 发送给所有的上级
List<ParentPlatform> parentPlatforms = storage.queryEnableParentPlatformList(true);
if (parentPlatforms.size() > 0) {
for (ParentPlatform parentPlatform : parentPlatforms) {
try {
commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 发送报警: {}", e.getMessage());
}
}
}
}else {
// 获取开启了消息推送的设备和平台
List<ParentPlatform> parentPlatforms = storage.queryEnablePlatformListWithAsMessageChannel();
if (parentPlatforms.size() > 0) {
for (ParentPlatform parentPlatform : parentPlatforms) {
try {
commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 发送报警: {}", e.getMessage());
}
}
}
}
// 获取开启了消息推送的设备和平台
List<Device> devices = storage.queryDeviceWithAsMessageChannel();
if (devices.size() > 0) {
for (Device device : devices) {
try {
commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 发送报警: {}", e.getMessage());
commander.sendAlarmMessage(device, deviceAlarm);
} catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 发送报警: {}", e.getMessage());
}
}
}
}else {
Device device = storage.queryVideoDevice(gbId);
ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);

3
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorage.java
View File

@@ -378,4 +378,7 @@ public interface IVideoManagerStorage {
List<DeviceChannelExtend> queryChannelsByDeviceId(String serial, List<String> channelIds, Boolean online);
List<ParentPlatform> queryEnablePlatformListWithAsMessageChannel();
List<Device> queryDeviceWithAsMessageChannel();
}

4
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
View File

@@ -344,10 +344,10 @@ public interface DeviceChannelMapper {
"select * " +
"from device_channel " +
"where deviceId=#{deviceId}" +
" <if test='parentId != null and length != null' > and parentId = #{parentId} or left(channelId, #{parentId.length()}) = #{parentId} and length(channelId)=#{length} </if>" +
" <if test='parentId != null and length != null' > and parentId = #{parentId} or left(channelId, LENGTH(#{parentId})) = #{parentId} and length(channelId)=#{length} </if>" +
" <if test='parentId == null and length != null' > and parentId = #{parentId} or length(channelId)=#{length} </if>" +
" <if test='parentId == null and length == null' > and parentId = #{parentId} </if>" +
" <if test='parentId != null and length == null' > and parentId = #{parentId} or left(channelId, #{parentId.length()}) = #{parentId} </if>" +
" <if test='parentId != null and length == null' > and parentId = #{parentId} or left(channelId, LENGTH(#{parentId})) = #{parentId} </if>" +
" </script>"})
List<DeviceChannel> getChannelsWithCivilCodeAndLength(String deviceId, String parentId, Integer length);

12
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java
View File

@@ -39,6 +39,7 @@ public interface DeviceMapper {
"mobilePositionSubmissionInterval," +
"subscribeCycleForAlarm," +
"ssrcCheck," +
"asMessageChannel," +
"geoCoordSys," +
"treeType," +
"online" +
@@ -70,6 +71,7 @@ public interface DeviceMapper {
"mobilePositionSubmissionInterval," +
"subscribeCycleForAlarm," +
"ssrcCheck," +
"asMessageChannel," +
"geoCoordSys," +
"treeType," +
"online" +
@@ -98,6 +100,7 @@ public interface DeviceMapper {
"#{mobilePositionSubmissionInterval}," +
"#{subscribeCycleForAlarm}," +
"#{ssrcCheck}," +
"#{asMessageChannel}," +
"#{geoCoordSys}," +
"#{treeType}," +
"#{online}" +
@@ -152,6 +155,7 @@ public interface DeviceMapper {
"mobilePositionSubmissionInterval," +
"subscribeCycleForAlarm," +
"ssrcCheck," +
"asMessageChannel," +
"geoCoordSys," +
"treeType," +
"online," +
@@ -192,6 +196,7 @@ public interface DeviceMapper {
"mobilePositionSubmissionInterval," +
"subscribeCycleForAlarm," +
"ssrcCheck," +
"asMessageChannel," +
"geoCoordSys," +
"treeType," +
"online " +
@@ -222,6 +227,7 @@ public interface DeviceMapper {
"mobilePositionSubmissionInterval," +
"subscribeCycleForAlarm," +
"ssrcCheck," +
"asMessageChannel," +
"geoCoordSys," +
"treeType," +
"online" +
@@ -243,6 +249,7 @@ public interface DeviceMapper {
"<if test=\"mobilePositionSubmissionInterval != null\">, mobilePositionSubmissionInterval=#{mobilePositionSubmissionInterval}</if>" +
"<if test=\"subscribeCycleForAlarm != null\">, subscribeCycleForAlarm=#{subscribeCycleForAlarm}</if>" +
"<if test=\"ssrcCheck != null\">, ssrcCheck=#{ssrcCheck}</if>" +
"<if test=\"asMessageChannel != null\">, asMessageChannel=#{asMessageChannel}</if>" +
"<if test=\"geoCoordSys != null\">, geoCoordSys=#{geoCoordSys}</if>" +
"<if test=\"treeType != null\">, treeType=#{treeType}</if>" +
"<if test=\"mediaServerId != null\">, mediaServerId=#{mediaServerId}</if>" +
@@ -259,6 +266,7 @@ public interface DeviceMapper {
"updateTime," +
"charset," +
"ssrcCheck," +
"asMessageChannel," +
"geoCoordSys," +
"treeType," +
"online" +
@@ -271,6 +279,7 @@ public interface DeviceMapper {
"#{updateTime}," +
"#{charset}," +
"#{ssrcCheck}," +
"#{asMessageChannel}," +
"#{geoCoordSys}," +
"#{treeType}," +
"#{online}" +
@@ -282,4 +291,7 @@ public interface DeviceMapper {
@Select("select * from device")
List<Device> getAll();
@Select("select * from device where asMessageChannel = 1")
List<Device> queryDeviceWithAsMessageChannel();
}

10
src/main/java/com/genersoft/iot/vmp/storager/dao/ParentPlatformMapper.java
View File

@@ -15,10 +15,10 @@ import java.util.List;
public interface ParentPlatformMapper {
@Insert("INSERT INTO parent_platform (enable, name, serverGBId, serverGBDomain, serverIP, serverPort, deviceGBId, deviceIp, " +
" devicePort, username, password, expires, keepTimeout, transport, characterSet, ptz, rtcp, " +
" devicePort, username, password, expires, keepTimeout, transport, characterSet, ptz, rtcp, asMessageChannel, " +
" status, startOfflinePush, catalogId, administrativeDivision, catalogGroup, createTime, updateTime, treeType) " +
" VALUES (#{enable}, #{name}, #{serverGBId}, #{serverGBDomain}, #{serverIP}, #{serverPort}, #{deviceGBId}, #{deviceIp}, " +
" #{devicePort}, #{username}, #{password}, #{expires}, #{keepTimeout}, #{transport}, #{characterSet}, #{ptz}, #{rtcp}, " +
" #{devicePort}, #{username}, #{password}, #{expires}, #{keepTimeout}, #{transport}, #{characterSet}, #{ptz}, #{rtcp}, #{asMessageChannel}, " +
" #{status}, #{startOfflinePush}, #{catalogId}, #{administrativeDivision}, #{catalogGroup}, #{createTime}, #{updateTime}, #{treeType})")
int addParentPlatform(ParentPlatform parentPlatform);
@@ -40,6 +40,7 @@ public interface ParentPlatformMapper {
"characterSet=#{characterSet}, " +
"ptz=#{ptz}, " +
"rtcp=#{rtcp}, " +
"asMessageChannel=#{asMessageChannel}, " +
"status=#{status}, " +
"startOfflinePush=#{startOfflinePush}, " +
"catalogGroup=#{catalogGroup}, " +
@@ -68,9 +69,12 @@ public interface ParentPlatformMapper {
"FROM parent_platform pp ")
List<ParentPlatform> getParentPlatformList();
@Select("SELECT * FROM parent_platform WHERE enable=#{enable}")
@Select("SELECT * FROM parent_platform WHERE enable=#{enable} ")
List<ParentPlatform> getEnableParentPlatformList(boolean enable);
@Select("SELECT * FROM parent_platform WHERE enable=1 and asMessageChannel = 1")
List<ParentPlatform> queryEnablePlatformListWithAsMessageChannel();
@Select("SELECT * FROM parent_platform WHERE serverGBId=#{platformGbId}")
ParentPlatform getParentPlatByServerGBId(String platformGbId);

12
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
View File

@@ -177,12 +177,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
@Override
public boolean startDownload(StreamInfo stream, String callId) {
boolean result;
String key=String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId);
if (stream.getProgress() == 1) {
result = RedisUtil.set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
logger.debug("添加下载缓存==已完成下载=》{}",key);
result = RedisUtil.set(key, stream);
}else {
result = RedisUtil.set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream, 60*60);
logger.debug("添加下载缓存==未完成下载=》{}",key);
result = RedisUtil.set(key, stream, 60*60);
}
return result;
}
@@ -617,7 +619,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
stream,
callId
);
List<Object> streamInfoScan = RedisUtil.scan(key);
List<Object> streamInfoScan = RedisUtil.scan2(key);
if (streamInfoScan.size() > 0) {
return (StreamInfo) RedisUtil.get((String) streamInfoScan.get(0));
}else {

10
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
View File

@@ -524,6 +524,16 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return platformMapper.getEnableParentPlatformList(enable);
}
@Override
public List<ParentPlatform> queryEnablePlatformListWithAsMessageChannel() {
return platformMapper.queryEnablePlatformListWithAsMessageChannel();
}
@Override
public List<Device> queryDeviceWithAsMessageChannel() {
return deviceMapper.queryDeviceWithAsMessageChannel();
}
@Override
public void outlineForAllParentPlatform() {
platformMapper.outlineForAllParentPlatform();

8
src/main/java/com/genersoft/iot/vmp/utils/redis/RedisUtil.java
View File

@@ -881,7 +881,13 @@ public class RedisUtil {
return new ArrayList<>(resultKeys);
}
public static List<Object> scan2(String query) {
if (redisTemplate == null) {
redisTemplate = SpringBeanFactory.getBean("redisTemplate");
}
Set<String> keys = redisTemplate.keys(query);
return new ArrayList<>(keys);
}
// ============================== 消息发送与订阅 ==============================
public static void convertAndSend(String channel, JSONObject msg) {
if (redisTemplate == null) {