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

Merge remote-tracking branch 'origin/dev/master-通道结构重构' into dev/master-通道结构重构

Browse Source
This commit is contained in:
648540858
2024-09-28 12:48:05 +08:00
parent 55001e42d6 726380cec4
commit b4688f53dc
17 changed files with 171 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/genersoft/iot/vmp/gb28181/bean/GroupTree.java
View File

@@ -10,7 +10,7 @@ import lombok.Data;
@Schema(description = "业务分组树")
public class GroupTree extends Group{
@Schema(description = "是否有子节点")
@Schema(description = "树节点ID")
private String treeId;
@Schema(description = "是否有子节点")

3
src/main/java/com/genersoft/iot/vmp/gb28181/bean/RegionTree.java
View File

@@ -10,6 +10,9 @@ import lombok.Data;
@Schema(description = "区域树")
public class RegionTree extends Region {
@Schema(description = "树节点ID")
private String treeId;
@Schema(description = "是否有子节点")
private boolean isLeaf;

2
src/main/java/com/genersoft/iot/vmp/gb28181/dao/CommonGBChannelMapper.java
View File

@@ -263,6 +263,7 @@ public interface CommonGBChannelMapper {
@Select("<script>" +
" select " +
" id," +
" concat('channel', id) as tree_id," +
" coalesce(gb_device_id, device_id) as device_id," +
" coalesce(gb_name, name) as name, " +
" coalesce(gb_parent_id, parent_id) as parent_device_id, " +
@@ -352,6 +353,7 @@ public interface CommonGBChannelMapper {
@Select("<script>" +
" select " +
" id," +
" concat('channel', id) as tree_id," +
" coalesce(gb_device_id, device_id) as device_id," +
" coalesce(gb_name, name) as name, " +
" coalesce(gb_parent_id, parent_id) as parent_device_id, " +

12
src/main/java/com/genersoft/iot/vmp/gb28181/dao/DeviceChannelMapper.java
View File

@@ -688,18 +688,10 @@ public interface DeviceChannelMapper {
@Update({"<script>" +
"<foreach collection='channels' item='item' separator=';'>" +
"UPDATE wvp_device_channel SET status='ON' WHERE device_id=#{item.deviceId}" +
"UPDATE wvp_device_channel SET status=#{item.status} WHERE device_id=#{item.deviceId}" +
"</foreach>" +
"</script>"})
int batchOnlineForNotify(List<DeviceChannel> channels);
@Update({"<script>" +
"<foreach collection='channels' item='item' separator=';'>" +
"UPDATE wvp_device_channel SET status='OFF' WHERE device_id=#{item.deviceId}" +
"</foreach>" +
"</script>"})
int batchOfflineForNotify(List<DeviceChannel> channels);
int batchUpdateStatus(List<DeviceChannel> channels);
@Select("select count(1) from wvp_device_channel where status = 'ON'")
int getOnlineCount();

1
src/main/java/com/genersoft/iot/vmp/gb28181/dao/GroupMapper.java
View File

@@ -67,6 +67,7 @@ public interface GroupMapper {
@Select(" <script>" +
" SELECT " +
" * , " +
" concat('group', id) as tree_id," +
" 0 as type," +
" false as is_leaf" +
" from wvp_common_group " +

1
src/main/java/com/genersoft/iot/vmp/gb28181/dao/RegionMapper.java
View File

@@ -72,6 +72,7 @@ public interface RegionMapper {
@Select(" <script>" +
" SELECT " +
" *, " +
" concat('region', id) as tree_id," +
" 0 as type," +
" false as is_leaf" +
" from wvp_common_region " +

4
src/main/java/com/genersoft/iot/vmp/gb28181/dao/provider/ChannelProvider.java
View File

@@ -96,10 +96,10 @@ public class ChannelProvider {
sqlBuild.append(" AND coalesce(gb_civil_code, civil_code) is null");
}
if (params.get("hasGroup") != null && (Boolean)params.get("hasGroup")) {
sqlBuild.append(" AND coalesce(gb_business_group_id, business_group_id) is not null");
sqlBuild.append(" AND coalesce(gb_parent_id, parent_id) is not null");
}
if (params.get("hasGroup") != null && !(Boolean)params.get("hasGroup")) {
sqlBuild.append(" AND coalesce(gb_business_group_id, business_group_id) is null");
sqlBuild.append(" AND coalesce(gb_parent_id, parent_id) is null");
}
return sqlBuild.toString();
}

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

@@ -3,8 +3,8 @@ package com.genersoft.iot.vmp.gb28181.service;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.gb28181.controller.bean.ChannelReduce;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.web.gb28181.dto.DeviceChannelExtend;
import com.github.pagehelper.PageInfo;
@@ -47,15 +47,7 @@ public interface IDeviceChannelService {
*/
int deleteChannelsForNotify(List<DeviceChannel> deleteChannelList);
/**
* 批量上线
*/
int channelsOnlineForNotify(List<DeviceChannel> channels);
/**
* 批量下线
*/
int channelsOfflineForNotify(List<DeviceChannel> channels);
int updateChannelsStatus(List<DeviceChannel> channels);
/**
* 获取一个通道

70
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceChannelServiceImpl.java
View File

@@ -193,13 +193,45 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
@Override
public int deleteChannelsForNotify(List<DeviceChannel> deleteChannelList) {
return channelMapper.batchDelForNotify(deleteChannelList);
@Transactional
public int deleteChannelsForNotify(List<DeviceChannel> channels) {
int limitCount = 1000;
int result = 0;
if (!channels.isEmpty()) {
if (channels.size() > limitCount) {
for (int i = 0; i < channels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > channels.size()) {
toIndex = channels.size();
}
result += channelMapper.batchDel(channels.subList(i, toIndex));
}
}else {
result += channelMapper.batchDel(channels);
}
}
return result;
}
@Transactional
@Override
public int channelsOnlineForNotify(List<DeviceChannel> channels) {
return channelMapper.batchOnlineForNotify(channels);
public int updateChannelsStatus(List<DeviceChannel> channels) {
int limitCount = 1000;
int result = 0;
if (!channels.isEmpty()) {
if (channels.size() > limitCount) {
for (int i = 0; i < channels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > channels.size()) {
toIndex = channels.size();
}
result += channelMapper.batchUpdateStatus(channels.subList(i, toIndex));
}
}else {
result += channelMapper.batchUpdateStatus(channels);
}
}
return result;
}
@Override
@@ -207,12 +239,6 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
channelMapper.online(channel.getId());
}
@Override
public int channelsOfflineForNotify(List<DeviceChannel> channels) {
return channelMapper.batchOfflineForNotify(channels);
}
@Override
public void offline(DeviceChannel channel) {
channelMapper.offline(channel.getId());
@@ -242,6 +268,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
@Override
@Transactional
public synchronized void batchUpdateChannelForNotify(List<DeviceChannel> channels) {
String now = DateUtil.getNow();
for (DeviceChannel channel : channels) {
@@ -264,8 +291,27 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
@Override
@Transactional
public void batchAddChannel(List<DeviceChannel> channels) {
channelMapper.batchAdd(channels);
String now = DateUtil.getNow();
for (DeviceChannel channel : channels) {
channel.setUpdateTime(now);
channel.setCreateTime(now);
}
int limitCount = 1000;
if (!channels.isEmpty()) {
if (channels.size() > limitCount) {
for (int i = 0; i < channels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > channels.size()) {
toIndex = channels.size();
}
channelMapper.batchAdd(channels.subList(i, toIndex));
}
}else {
channelMapper.batchAdd(channels);
}
}
for (DeviceChannel channel : channels) {
if (channel.getParentId() != null) {
channelMapper.updateChannelSubCount(channel.getDeviceDbId(), channel.getParentId());
@@ -340,7 +386,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", mobilePosition.getDeviceId());
jsonObject.put("code", mobilePosition.getChannelId());
jsonObject.put("code", channel.getDeviceId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());

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

@@ -39,8 +39,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
@Component
public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent {
private final List<DeviceChannel> updateChannelOnlineList = new CopyOnWriteArrayList<>();
private final List<DeviceChannel> updateChannelOfflineList = new CopyOnWriteArrayList<>();
private final List<DeviceChannel> updateChannelForStatusChange = new CopyOnWriteArrayList<>();
private final Map<String, DeviceChannel> updateChannelMap = new ConcurrentHashMap<>();
private final Map<String, DeviceChannel> addChannelMap = new ConcurrentHashMap<>();
@@ -60,6 +59,10 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
@Autowired
private IDeviceChannelService deviceChannelService;
// @Scheduled(fixedRate = 2000) //每400毫秒执行一次
// public void showSize(){
// log.warn("[notify-目录订阅] 待处理消息数量： {}", taskQueue.size() );
// }
@Transactional
public void process(RequestEvent evt) {
@@ -75,7 +78,14 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
if (taskQueue.isEmpty()) {
return;
}
for (HandlerCatchData take : taskQueue) {
List<HandlerCatchData> handlerCatchDataList = new ArrayList<>();
while (!taskQueue.isEmpty()) {
handlerCatchDataList.add(taskQueue.poll());
}
if (handlerCatchDataList.isEmpty()) {
return;
}
for (HandlerCatchData take : handlerCatchDataList) {
if (take == null) {
continue;
}
@@ -119,14 +129,17 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
log.error("[解析CatalogChannelEvent]失败原文: \n{}", new String(evt.getRequest().getRawContent(), Charset.forName(device.getCharset())));
continue;
}
log.info("[收到目录订阅]：{}/{}-{}", device.getDeviceId(),
catalogChannelEvent.getChannel().getDeviceId(), catalogChannelEvent.getEvent());
if (log.isDebugEnabled()){
log.debug("[收到目录订阅]：{}/{}-{}", device.getDeviceId(),
catalogChannelEvent.getChannel().getDeviceId(), catalogChannelEvent.getEvent());
}
DeviceChannel channel = catalogChannelEvent.getChannel();
switch (catalogChannelEvent.getEvent()) {
case CatalogEvent.ON:
// 上线
log.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
updateChannelOnlineList.add(catalogChannelEvent.getChannel());
channel.setStatus("ON");
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), true);
@@ -138,7 +151,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
log.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
} else {
updateChannelOfflineList.add(catalogChannelEvent.getChannel());
channel.setStatus("OFF");
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
@@ -151,7 +165,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
log.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
} else {
updateChannelOfflineList.add(catalogChannelEvent.getChannel());
channel.setStatus("OFF");
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
@@ -164,7 +179,8 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
log.info("[收到通道视频故障通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
} else {
updateChannelOfflineList.add(catalogChannelEvent.getChannel());
channel.setStatus("OFF");
updateChannelForStatusChange.add(channel);
if (userSetting.getDeviceStatusNotify()) {
// 发送redis消息
redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId(), false);
@@ -178,7 +194,6 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
DeviceChannel deviceChannel = deviceChannelService.getOne(deviceId, catalogChannelEvent.getChannel().getDeviceId());
if (deviceChannel != null) {
log.info("[增加通道] 已存在，不发送通知只更新，设备: {}, 通道 {}", device.getDeviceId(), catalogChannelEvent.getChannel().getDeviceId());
DeviceChannel channel = catalogChannelEvent.getChannel();
channel.setId(deviceChannel.getId());
channel.setHasAudio(deviceChannel.isHasAudio());
channel.setUpdateTime(DateUtil.getNow());
@@ -210,7 +225,6 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
// 判断此通道是否存在
DeviceChannel deviceChannelForUpdate = deviceChannelService.getOne(deviceId, catalogChannelEvent.getChannel().getDeviceId());
if (deviceChannelForUpdate != null) {
DeviceChannel channel = catalogChannelEvent.getChannel();
channel.setId(deviceChannelForUpdate.getId());
channel.setHasAudio(deviceChannelForUpdate.isHasAudio());
channel.setUpdateTime(DateUtil.getNow());
@@ -242,8 +256,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
taskQueue.clear();
if (!updateChannelMap.keySet().isEmpty()
|| !addChannelMap.keySet().isEmpty()
|| !updateChannelOnlineList.isEmpty()
|| !updateChannelOfflineList.isEmpty()
|| !updateChannelForStatusChange.isEmpty()
|| !deleteChannelList.isEmpty()) {
executeSave();
}
@@ -256,14 +269,9 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
log.error("[存储收到的增加通道] 异常： ", e );
}
try {
executeSaveForOnline();
executeSaveForStatus();
} catch (Exception e) {
log.error("[存储收到的通道上线] 异常： ", e );
}
try {
executeSaveForOffline();
} catch (Exception e) {
log.error("[存储收到的通道离线] 异常： ", e );
log.error("[存储收到的通道状态变化] 异常： ", e );
}
try {
executeSaveForUpdate();
@@ -301,17 +309,10 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
}
}
private void executeSaveForOnline(){
if (!updateChannelOnlineList.isEmpty()) {
deviceChannelService.channelsOnlineForNotify(updateChannelOnlineList);
updateChannelOnlineList.clear();
}
}
private void executeSaveForOffline(){
if (!updateChannelOfflineList.isEmpty()) {
deviceChannelService.channelsOfflineForNotify(updateChannelOfflineList);
updateChannelOfflineList.clear();
private void executeSaveForStatus(){
if (!updateChannelForStatusChange.isEmpty()) {
deviceChannelService.updateChannelsStatus(updateChannelForStatusChange);
updateChannelForStatusChange.clear();
}
}
}

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

@@ -24,6 +24,7 @@ import org.springframework.util.ObjectUtils;
import javax.sip.RequestEvent;
import javax.sip.header.FromHeader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -65,7 +66,14 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
if (taskQueue.isEmpty()) {
return;
}
for (HandlerCatchData take : taskQueue) {
List<HandlerCatchData> handlerCatchDataList = new ArrayList<>();
while (!taskQueue.isEmpty()) {
handlerCatchDataList.add(taskQueue.poll());
}
if (handlerCatchDataList.isEmpty()) {
return;
}
for (HandlerCatchData take : handlerCatchDataList) {
if (take == null) {
continue;
}
@@ -89,13 +97,15 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
mobilePosition.setDeviceId(device.getDeviceId());
mobilePosition.setDeviceName(device.getName());
mobilePosition.setCreateTime(DateUtil.getNow());
DeviceChannel deviceChannel = null;
List<Element> elements = rootElement.elements();
for (Element element : elements) {
switch (element.getName()){
case "DeviceID":
String channelId = element.getStringValue();
if (!deviceId.equals(channelId)) {
DeviceChannel deviceChannel = deviceChannelService.getOne(device.getDeviceId(), channelId);
deviceChannel = deviceChannelService.getOne(device.getDeviceId(), channelId);
if (deviceChannel != null) {
mobilePosition.setChannelId(deviceChannel.getId());
}
@@ -154,13 +164,13 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
}catch (Exception e) {
log.error("[向上级转发移动位置失败] ", e);
}
if (mobilePosition.getChannelId() == null || mobilePosition.getChannelId().equals(mobilePosition.getDeviceId())) {
if (mobilePosition.getChannelId() == null) {
List<DeviceChannel> channels = deviceChannelService.queryChaneListByDeviceId(mobilePosition.getDeviceId());
channels.forEach(channel -> {
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", channel.getDeviceId());
jsonObject.put("serial", device.getDeviceId());
jsonObject.put("code", channel.getDeviceId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
@@ -171,16 +181,18 @@ public class NotifyRequestForMobilePositionProcessor extends SIPRequestProcessor
});
}else {
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", mobilePosition.getDeviceId());
jsonObject.put("code", mobilePosition.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
if (deviceChannel != null) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", mobilePosition.getDeviceId());
jsonObject.put("code", deviceChannel.getDeviceId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
}
}
} catch (DocumentException e) {
log.error("未处理的异常 ", e);

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

@@ -29,7 +29,6 @@ public class RedisPushStreamResponseListener implements MessageListener {
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
private Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
public interface PushStreamResponseEvent{

2
src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java
View File

@@ -241,7 +241,7 @@ public class RedisRpcController {
return response;
}
MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream());
if (mediaInfo != null) {
if (mediaInfo == null) {
log.info("[redis-rpc] startSendRtp->流不在线： {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() );
WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "流不在线");
response.setBody(wvpResult);