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

通道编辑修复分组路径显示异常，支持行政区划路径显示

Browse Source
This commit is contained in:
lin
2025-11-19 17:03:45 +08:00
parent c1c9c7957b
commit 033db1925c
21 changed files with 279 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
src/main/java/com/genersoft/iot/vmp/gb28181/bean/CommonGBChannel.java
View File

@@ -396,4 +396,56 @@ public class CommonGBChannel {
return commonGBChannel;
}
@Override
public String toString() {
return "CommonGBChannel{" +
"gbId=" + gbId +
", gbDeviceId='" + gbDeviceId + '\'' +
", gbName='" + gbName + '\'' +
", gbManufacturer='" + gbManufacturer + '\'' +
", gbModel='" + gbModel + '\'' +
", gbOwner='" + gbOwner + '\'' +
", gbCivilCode='" + gbCivilCode + '\'' +
", gbBlock='" + gbBlock + '\'' +
", gbAddress='" + gbAddress + '\'' +
", gbParental=" + gbParental +
", gbParentId='" + gbParentId + '\'' +
", gbSafetyWay=" + gbSafetyWay +
", gbRegisterWay=" + gbRegisterWay +
", gbCertNum='" + gbCertNum + '\'' +
", gbCertifiable=" + gbCertifiable +
", gbErrCode=" + gbErrCode +
", gbEndTime='" + gbEndTime + '\'' +
", gbSecrecy=" + gbSecrecy +
", gbIpAddress='" + gbIpAddress + '\'' +
", gbPort=" + gbPort +
", gbPassword='" + gbPassword + '\'' +
", gbStatus='" + gbStatus + '\'' +
", gbLongitude=" + gbLongitude +
", gbLatitude=" + gbLatitude +
", gpsAltitude=" + gpsAltitude +
", gpsSpeed=" + gpsSpeed +
", gpsDirection=" + gpsDirection +
", gpsTime='" + gpsTime + '\'' +
", gbBusinessGroupId='" + gbBusinessGroupId + '\'' +
", gbPtzType=" + gbPtzType +
", gbPositionType=" + gbPositionType +
", gbRoomType=" + gbRoomType +
", gbUseType=" + gbUseType +
", gbSupplyLightType=" + gbSupplyLightType +
", gbDirectionType=" + gbDirectionType +
", gbResolution='" + gbResolution + '\'' +
", gbDownloadSpeed='" + gbDownloadSpeed + '\'' +
", gbSvcSpaceSupportMod=" + gbSvcSpaceSupportMod +
", gbSvcTimeSupportMode=" + gbSvcTimeSupportMode +
", recordPLan=" + recordPLan +
", dataType=" + dataType +
", dataDeviceId=" + dataDeviceId +
", createTime='" + createTime + '\'' +
", updateTime='" + updateTime + '\'' +
", streamId='" + streamId + '\'' +
", enableBroadcast=" + enableBroadcast +
", mapLevel=" + mapLevel +
'}';
}
}

18
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java
View File

@@ -32,7 +32,8 @@ public class SubscribeHolder {
public void putCatalogSubscribe(String platformId, SubscribeInfo subscribeInfo) {
log.info("[国标级联] 添加目录订阅，平台： {}， 有效期： {}", platformId, subscribeInfo.getExpires());
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "catalog", platformId);
subscribeInfo.setServerId(userSetting.getServerId());
String key = String.format("%s:%s:%s", prefix, "catalog", platformId);
if (subscribeInfo.getExpires() > 0) {
Duration duration = Duration.ofSeconds(subscribeInfo.getExpires());
redisTemplate.opsForValue().set(key, subscribeInfo, duration);
@@ -42,18 +43,19 @@ public class SubscribeHolder {
}
public SubscribeInfo getCatalogSubscribe(String platformId) {
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "catalog", platformId);
String key = String.format("%s:%s:%s", prefix, "catalog", platformId);
return (SubscribeInfo)redisTemplate.opsForValue().get(key);
}
public void removeCatalogSubscribe(String platformId) {
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "catalog", platformId);
String key = String.format("%s:%s:%s", prefix, "catalog", platformId);
redisTemplate.delete(key);
}
public void putMobilePositionSubscribe(String platformId, SubscribeInfo subscribeInfo, Runnable gpsTask) {
log.info("[国标级联] 添加移动位置订阅，平台： {}， 有效期： {}s", platformId, subscribeInfo.getExpires());
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "mobilePosition", platformId);
subscribeInfo.setServerId(userSetting.getServerId());
String key = String.format("%s:%s:%s", prefix, "mobilePosition", platformId);
if (subscribeInfo.getExpires() > 0) {
Duration duration = Duration.ofSeconds(subscribeInfo.getExpires());
redisTemplate.opsForValue().set(key, subscribeInfo, duration);
@@ -81,12 +83,12 @@ public class SubscribeHolder {
}
public SubscribeInfo getMobilePositionSubscribe(String platformId) {
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "mobilePosition", platformId);
String key = String.format("%s:%s:%s", prefix, "mobilePosition", platformId);
return (SubscribeInfo)redisTemplate.opsForValue().get(key);
}
public void removeMobilePositionSubscribe(String platformId) {
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "mobilePosition", platformId);
String key = String.format("%s:%s:%s", prefix, "mobilePosition", platformId);
redisTemplate.delete(key);
}
@@ -96,7 +98,7 @@ public class SubscribeHolder {
}
List<String> result = new ArrayList<>();
for (Platform platform : platformList) {
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "catalog", platform.getServerGBId());
String key = String.format("%s:%s:%s", prefix, "catalog", platform.getServerGBId());
if (redisTemplate.hasKey(key)) {
result.add(platform.getServerGBId());
}
@@ -110,7 +112,7 @@ public class SubscribeHolder {
}
List<String> result = new ArrayList<>();
for (Platform platform : platformList) {
String key = String.format("%s_%s_%s_%s", prefix, userSetting.getServerId(), "mobilePosition", platform.getServerGBId());
String key = String.format("%s:%s:%s", prefix, "mobilePosition", platform.getServerGBId());
if (redisTemplate.hasKey(key)) {
result.add(platform.getServerGBId());
}

5
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java
View File

@@ -37,6 +37,11 @@ public class SubscribeInfo {
*/
private String simulatedCallId;
/**
* 来源serverId
*/
private String serverId;
public static SubscribeInfo getInstance(SIPResponse response, String id, int expires, EventHeader eventHeader){
SubscribeInfo subscribeInfo = new SubscribeInfo();

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

@@ -22,13 +22,13 @@ public interface DeviceChannelMapper {
"insert into wvp_device_channel " +
"(device_id, data_type, data_device_id, name, manufacturer, model, owner, civil_code, block, " +
"address, parental, parent_id, safety_way, register_way, cert_num, certifiable, err_code, end_time, secrecy, " +
"ip_address, port, password, status, longitude, latitude, gb_longitude, gb_latitude, ptz_type, position_type, room_type, use_type, " +
"ip_address, port, password, status, longitude, latitude, ptz_type, position_type, room_type, use_type, " +
"supply_light_type, direction_type, resolution, business_group_id, download_speed, svc_space_support_mod, " +
"svc_time_support_mode, create_time, update_time, sub_count, stream_id, has_audio, gps_time, stream_identification, channel_type) " +
"values " +
"(#{deviceId}, #{dataType}, #{dataDeviceId}, #{name}, #{manufacturer}, #{model}, #{owner}, #{civilCode}, #{block}, " +
"#{address}, #{parental}, #{parentId}, #{safetyWay}, #{registerWay}, #{certNum}, #{certifiable}, #{errCode}, #{endTime}, #{secrecy}, " +
"#{ipAddress}, #{port}, #{password}, #{status}, #{longitude}, #{latitude}, #{gbLongitude}, #{gbLatitude}, #{ptzType}, #{positionType}, #{roomType}, #{useType}, " +
"#{ipAddress}, #{port}, #{password}, #{status}, #{longitude}, #{latitude}, #{ptzType}, #{positionType}, #{roomType}, #{useType}, " +
"#{supplyLightType}, #{directionType}, #{resolution}, #{businessGroupId}, #{downloadSpeed}, #{svcSpaceSupportMod}," +
" #{svcTimeSupportMode}, #{createTime}, #{updateTime}, #{subCount}, #{streamId}, #{hasAudio}, #{gpsTime}, #{streamIdentification}, #{channelType}) " +
"</script>")
@@ -63,8 +63,6 @@ public interface DeviceChannelMapper {
", status=#{status}" +
", longitude=#{longitude}" +
", latitude=#{latitude}" +
", gb_longitude=#{gbLongitude}" +
", gb_latitude=#{gbLatitude}" +
", ptz_type=#{ptzType}" +
", position_type=#{positionType}" +
", room_type=#{roomType}" +
@@ -200,14 +198,14 @@ public interface DeviceChannelMapper {
"insert into wvp_device_channel " +
"(device_id, data_type, data_device_id, name, manufacturer, model, owner, civil_code, block, " +
"address, parental, parent_id, safety_way, register_way, cert_num, certifiable, err_code, end_time, secrecy, " +
"ip_address, port, password, status, longitude, latitude, gb_longitude, gb_latitude, ptz_type, position_type, room_type, use_type, " +
"ip_address, port, password, status, longitude, latitude, ptz_type, position_type, room_type, use_type, " +
"supply_light_type, direction_type, resolution, business_group_id, download_speed, svc_space_support_mod, " +
"svc_time_support_mode, create_time, update_time, sub_count, stream_id, has_audio, gps_time, stream_identification, channel_type) " +
"values " +
"<foreach collection='addChannels' index='index' item='item' separator=','> " +
"(#{item.deviceId}, #{item.dataType}, #{item.dataDeviceId}, #{item.name}, #{item.manufacturer}, #{item.model}, #{item.owner}, #{item.civilCode}, #{item.block}, " +
"#{item.address}, #{item.parental}, #{item.parentId}, #{item.safetyWay}, #{item.registerWay}, #{item.certNum}, #{item.certifiable}, #{item.errCode}, #{item.endTime}, #{item.secrecy}, " +
"#{item.ipAddress}, #{item.port}, #{item.password}, #{item.status}, #{item.longitude}, #{item.latitude}, #{item.gbLongitude}, #{item.gbLatitude}, #{item.ptzType}, #{item.positionType}, #{item.roomType}, #{item.useType}, " +
"#{item.ipAddress}, #{item.port}, #{item.password}, #{item.status}, #{item.longitude}, #{item.latitude}, #{item.ptzType}, #{item.positionType}, #{item.roomType}, #{item.useType}, " +
"#{item.supplyLightType}, #{item.directionType}, #{item.resolution}, #{item.businessGroupId}, #{item.downloadSpeed}, #{item.svcSpaceSupportMod}," +
" #{item.svcTimeSupportMode}, #{item.createTime}, #{item.updateTime}, #{item.subCount}, #{item.streamId}, #{item.hasAudio}, #{item.gpsTime}, #{item.streamIdentification}, #{item.channelType}) " +
"</foreach> " +
@@ -534,8 +532,6 @@ public interface DeviceChannelMapper {
", status=#{status}" +
", longitude=#{longitude}" +
", latitude=#{latitude}" +
", gb_longitude=#{gbLongitude}" +
", gb_latitude=#{gbLatitude}" +
", ptz_type=#{ptzType}" +
", position_type=#{positionType}" +
", room_type=#{roomType}" +

3
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
View File

@@ -65,16 +65,19 @@ public class EventPublisher {
}
public void channelEventPublishForUpdate(CommonGBChannel commonGBChannel, CommonGBChannel deviceChannelForOld) {
log.info("[通道改变内部分发-更新] {}", commonGBChannel.getGbDeviceId());
ChannelEvent channelEvent = ChannelEvent.getInstanceForUpdate(this, Collections.singletonList(commonGBChannel), Collections.singletonList(deviceChannelForOld));
applicationEventPublisher.publishEvent(channelEvent);
}
public void channelEventPublishForUpdate(List<CommonGBChannel> channelList, List<CommonGBChannel> channelListForOld) {
log.info("[通道改变内部分发-更新] 数量： {}", channelList.size());
ChannelEvent channelEvent = ChannelEvent.getInstanceForUpdate(this, channelList, channelListForOld);
applicationEventPublisher.publishEvent(channelEvent);
}
public void channelEventPublish(List<CommonGBChannel> channelList, ChannelEvent.ChannelEventMessageType type) {
log.info("[通道改变内部分发-{}] 数量： {}", type, channelList.size());
ChannelEvent channelEvent = ChannelEvent.getInstance(this, type, channelList);
applicationEventPublisher.publishEvent(channelEvent);
}

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

@@ -24,11 +24,11 @@ public interface IGbChannelService {
int offline(CommonGBChannel commonGBChannel);
int offline(List<CommonGBChannel> commonGBChannelList);
int offline(List<CommonGBChannel> commonGBChannelList, boolean permission);
int online(CommonGBChannel commonGBChannel);
int online(List<CommonGBChannel> commonGBChannelList);
int online(List<CommonGBChannel> commonGBChannelList, boolean permission);
void batchAdd(List<CommonGBChannel> commonGBChannels);
@@ -78,7 +78,7 @@ public interface IGbChannelService {
void deleteChannelToGroupByGbDevice(List<Integer> deviceIds);
void batchUpdate(List<CommonGBChannel> commonGBChannels);
void batchUpdateForStreamPushRedisMsg(List<CommonGBChannel> commonGBChannels, boolean permission);
CommonGBChannel queryOneWithPlatform(Integer platformId, String channelDeviceId);

89
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelServiceImpl.java
View File

@@ -187,7 +187,7 @@ public class GbChannelServiceImpl implements IGbChannelService, CommandLineRunne
@Override
public int update(CommonGBChannel commonGBChannel) {
log.info("[更新通道] 通道ID: {}, ", commonGBChannel.getGbId());
log.info("[更新通道] 通道ID: {}, ", commonGBChannel.toString());
if (commonGBChannel.getGbId() <= 0) {
log.warn("[更新通道] 未找到数据库ID，更新失败， {}({})", commonGBChannel.getGbName(), commonGBChannel.getGbDeviceId());
return 0;
@@ -248,32 +248,33 @@ public class GbChannelServiceImpl implements IGbChannelService, CommandLineRunne
@Override
@Transactional
public int offline(List<CommonGBChannel> commonGBChannelList) {
public int offline(List<CommonGBChannel> commonGBChannelList, boolean permission) {
if (commonGBChannelList.isEmpty()) {
log.warn("[多个通道离线] 通道数量为0，更新失败");
return 0;
}
log.info("[通道离线] 共 {} 个", commonGBChannelList.size());
int limitCount = 1000;
int result = 0;
if (commonGBChannelList.size() > limitCount) {
for (int i = 0; i < commonGBChannelList.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > commonGBChannelList.size()) {
toIndex = commonGBChannelList.size();
if (permission) {
int limitCount = 1000;
if (commonGBChannelList.size() > limitCount) {
for (int i = 0; i < commonGBChannelList.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > commonGBChannelList.size()) {
toIndex = commonGBChannelList.size();
}
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList.subList(i, toIndex), "OFF");
}
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList.subList(i, toIndex), "OFF");
} else {
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList, "OFF");
}
} else {
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList, "OFF");
log.info("[通道离线] 保存入库 共 {} 个改变", result);
}
if (result > 0) {
try {
// 发送catalog
eventPublisher.channelEventPublish(commonGBChannelList, ChannelEvent.ChannelEventMessageType.OFF);
} catch (Exception e) {
log.warn("[多个通道离线] 发送失败，数量：{}", commonGBChannelList.size(), e);
}
try {
// 发送catalog
eventPublisher.channelEventPublish(commonGBChannelList, ChannelEvent.ChannelEventMessageType.OFF);
} catch (Exception e) {
log.warn("[多个通道离线] 发送失败，数量：{}", commonGBChannelList.size(), e);
}
return result;
}
@@ -298,24 +299,26 @@ public class GbChannelServiceImpl implements IGbChannelService, CommandLineRunne
@Override
@Transactional
public int online(List<CommonGBChannel> commonGBChannelList) {
public int online(List<CommonGBChannel> commonGBChannelList, boolean permission) {
if (commonGBChannelList.isEmpty()) {
log.warn("[多个通道上线] 通道数量为0，更新失败");
return 0;
}
// 批量更新
int limitCount = 1000;
int result = 0;
if (commonGBChannelList.size() > limitCount) {
for (int i = 0; i < commonGBChannelList.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > commonGBChannelList.size()) {
toIndex = commonGBChannelList.size();
if (permission) {
// 批量更新
int limitCount = 1000;
if (commonGBChannelList.size() > limitCount) {
for (int i = 0; i < commonGBChannelList.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > commonGBChannelList.size()) {
toIndex = commonGBChannelList.size();
}
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList.subList(i, toIndex), "ON");
}
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList.subList(i, toIndex), "ON");
} else {
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList, "ON");
}
} else {
result += commonGBChannelMapper.updateStatusForListById(commonGBChannelList, "ON");
}
try {
// 发送catalog
@@ -358,27 +361,29 @@ public class GbChannelServiceImpl implements IGbChannelService, CommandLineRunne
}
@Override
public void batchUpdate(List<CommonGBChannel> commonGBChannels) {
public void batchUpdateForStreamPushRedisMsg(List<CommonGBChannel> commonGBChannels, boolean permission) {
if (commonGBChannels.isEmpty()) {
log.warn("[更新多个通道] 通道数量为0，更新失败");
return;
}
List<CommonGBChannel> oldCommonGBChannelList = commonGBChannelMapper.queryOldChanelListByChannels(commonGBChannels);
// 批量保存
int limitCount = 1000;
int result = 0;
if (commonGBChannels.size() > limitCount) {
for (int i = 0; i < commonGBChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > commonGBChannels.size()) {
toIndex = commonGBChannels.size();
if (permission) {
// 批量保存
int limitCount = 1000;
int result = 0;
if (commonGBChannels.size() > limitCount) {
for (int i = 0; i < commonGBChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > commonGBChannels.size()) {
toIndex = commonGBChannels.size();
}
result += commonGBChannelMapper.batchUpdate(commonGBChannels.subList(i, toIndex));
}
result += commonGBChannelMapper.batchUpdate(commonGBChannels.subList(i, toIndex));
} else {
result += commonGBChannelMapper.batchUpdate(commonGBChannels);
}
} else {
result += commonGBChannelMapper.batchUpdate(commonGBChannels);
log.info("[更新多个通道] 通道数量为{}，成功保存：{}", commonGBChannels.size(), result);
}
log.info("[更新多个通道] 通道数量为{}，成功保存：{}", commonGBChannels.size(), result);
// 发送通过更新通知
try {
// 发送通知

7
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GroupServiceImpl.java
View File

@@ -269,14 +269,12 @@ public class GroupServiceImpl implements IGroupService {
if (businessGroupInDb == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "业务分组不存在");
}
List<Group> groupList = new LinkedList<>();
groupList.add(businessGroupInDb);
Group group = groupManager.queryOneByDeviceId(deviceId, businessGroup);
if (group == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "虚拟组织不存在");
}
List<Group> allParent = getAllParent(group);
groupList.addAll(allParent);
List<Group> groupList = new LinkedList<>(allParent);
groupList.add(group);
return groupList;
}
@@ -286,10 +284,9 @@ public class GroupServiceImpl implements IGroupService {
return new ArrayList<>();
}
List<Group> groupList = new ArrayList<>();
Group parent = groupManager.queryOneByDeviceId(group.getParentDeviceId(), group.getBusinessGroup());
if (parent == null) {
return groupList;
return new ArrayList<>();
}
List<Group> allParent = getAllParent(parent);
allParent.add(parent);

13
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/RegionServiceImpl.java
View File

@@ -236,8 +236,9 @@ public class RegionServiceImpl implements IRegionService {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "行政区划不存在");
}
List<Region> allParent = getAllParent(region);
allParent.add(region);
return allParent;
List<Region> regionList = new LinkedList<>(allParent);
regionList.add(region);
return regionList;
}
@@ -246,15 +247,13 @@ public class RegionServiceImpl implements IRegionService {
return new ArrayList<>();
}
List<Region> regionList = new LinkedList<>();
Region parent = regionMapper.queryByDeviceId(region.getParentDeviceId());
if (parent == null) {
return regionList;
return new ArrayList<>();
}
regionList.add(parent);
List<Region> allParent = getAllParent(parent);
regionList.addAll(allParent);
return regionList;
allParent.add(parent);
return allParent;
}
@Override

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

@@ -86,7 +86,6 @@ public class MobilePositionServiceImpl implements IMobilePositionService {
/**
* 查询最新移动位置
* @param deviceId
*/
@Override
public MobilePosition queryLatestPosition(String deviceId) {
@@ -104,7 +103,7 @@ public class MobilePositionServiceImpl implements IMobilePositionService {
if (userSetting.getSavePositionHistory()) {
mobilePositionMapper.batchadd(mobilePositions);
}
log.debug("[移动位置订阅]更新通道位置： {}", mobilePositions.size());
log.info("[移动位置订阅]更新通道位置： {}", mobilePositions.size());
Map<String, Map<Integer, DeviceChannel>> updateChannelMap = new HashMap<>();
for (MobilePosition mobilePosition : mobilePositions) {
DeviceChannel deviceChannel = new DeviceChannel();

71
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGroupMsgListener.java
View File

@@ -109,50 +109,49 @@ public class RedisGroupMsgListener implements MessageListener {
continue;
}
Group topGroup = aliasGroupMap.get(groupMessage.getTopGroupGAlias());
if (topGroup == null) {
topGroup = aliasGroupToSave.get(groupMessage.getTopGroupGAlias());
}
if (topGroup == null) {
log.info("[REDIS消息-业务分组同步回复] 业务分组信息未发送或者未首先发送， {}", groupMessage.toString());
continue;
}
group.setBusinessGroup(topGroup.getDeviceId());
if (groupMessage.getParentGAlias() != null) {
Group parentGroup = aliasGroupMap.get(groupMessage.getParentGAlias());
if (parentGroup == null) {
parentGroup = aliasGroupToSave.get(groupMessage.getParentGAlias());
Group topGroup = aliasGroupMap.get(groupMessage.getTopGroupGAlias());
if (topGroup == null) {
topGroup = aliasGroupToSave.get(groupMessage.getTopGroupGAlias());
}
if (parentGroup == null) {
log.info("[REDIS消息-业务分组同步回复] 虚拟组织父节点未发送或者未首先发送， {}", groupMessage.toString());
if (topGroup == null) {
log.info("[REDIS消息-业务分组同步回复] 业务分组信息未发送或者未首先发送， {}", groupMessage.toString());
continue;
}
group.setParentId(null);
group.setParentDeviceId(parentGroup.getDeviceId());
group.setBusinessGroup(topGroup.getDeviceId());
if (groupMessage.getParentGAlias() != null) {
Group parentGroup = aliasGroupMap.get(groupMessage.getParentGAlias());
if (parentGroup == null) {
parentGroup = aliasGroupToSave.get(groupMessage.getParentGAlias());
}
if (parentGroup == null) {
log.info("[REDIS消息-业务分组同步回复] 虚拟组织父节点未发送或者未首先发送， {}", groupMessage.toString());
continue;
}
group.setParentId(null);
group.setParentDeviceId(parentGroup.getDeviceId());
} else {
group.setParentId(null);
group.setParentDeviceId(topGroup.getDeviceId());
}
} else {
group.setParentId(null);
group.setParentDeviceId(topGroup.getDeviceId());
group.setBusinessGroup(group.getDeviceId());
group.setParentDeviceId(null);
}
} else {
group.setParentId(null);
group.setBusinessGroup(group.getDeviceId());
group.setParentDeviceId(null);
group.setUpdateTime(DateUtil.getNow());
aliasGroupToSave.put(group.getAlias(), group);
}
group.setUpdateTime(DateUtil.getNow());
aliasGroupToSave.put(group.getAlias(), group);
log.info("[业务分组同步回复-存储分组数据] {}", JSONObject.toJSONString(aliasGroupToSave.values()));
// 存储分组数据
groupService.saveByAlias(aliasGroupToSave.values());
} catch (ControllerException e) {
log.warn("[REDIS消息-业务分组同步回复] 失败, \r\n{}", e.getMsg());
} catch (Exception e) {
log.warn("[REDIS消息-业务分组同步回复] 发现未处理的异常, \r\n{}", new String(msg.getBody()));
log.error("[REDIS消息-业务分组同步回复] 异常内容： ", e);
}
log.info("[业务分组同步回复-存储分组数据] {}", JSONObject.toJSONString(aliasGroupToSave.values()));
// 存储分组数据
groupService.saveByAlias(aliasGroupToSave.values());
} catch (ControllerException e) {
log.warn("[REDIS消息-业务分组同步回复] 失败, \r\n{}", e.getMsg());
} catch (Exception e) {
log.warn("[REDIS消息-业务分组同步回复] 发现未处理的异常, \r\n{}", new String(msg.getBody()));
log.error("[REDIS消息-业务分组同步回复] 异常内容： ", e);
}
}
}
@@ -175,7 +174,7 @@ public class RedisGroupMsgListener implements MessageListener {
codeType = "215";
}
return String.format(deviceTemplate, codeType, RandomStringUtils.insecure().next(6, false, true));
}catch (Exception e) {
} catch (Exception e) {
log.error("[REDIS消息-业务分组同步回复] 构建新的分组编号失败", e);
return null;
}

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

@@ -48,10 +48,6 @@ public class RedisPushStreamListMsgListener implements MessageListener {
@Override
public void onMessage(Message message, byte[] bytes) {
String serverId = redisCatchStorage.chooseOneServer(null);
if (!userSetting.getServerId().equals(serverId)) {
return;
}
log.info("[REDIS: 推流设备列表更新]： {}", new String(message.getBody()));
taskQueue.offer(message);
}
@@ -130,7 +126,7 @@ public class RedisPushStreamListMsgListener implements MessageListener {
if (!streamPushItemForUpdate.isEmpty()) {
log.info("修改{}条", streamPushItemForUpdate.size());
log.info(JSONObject.toJSONString(streamPushItemForUpdate));
streamPushService.batchUpdate(streamPushItemForUpdate);
streamPushService.batchUpdateForRedisMsg(streamPushItemForUpdate);
}
} catch (Exception e) {
log.warn("[REDIS消息-推流设备列表更新] 发现未处理的异常, \r\n{}", new String(msg.getBody()));

8
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java
View File

@@ -79,19 +79,19 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
dynamicTask.stop(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED);
if (streamStatusMessage.isSetAllOffline()) {
// 所有设备离线
streamPushService.allOffline();
streamPushService.allOfflineForRedisMsg();
}
if (streamStatusMessage.getOfflineStreams() != null
&& !streamStatusMessage.getOfflineStreams().isEmpty()) {
// 更新部分设备离线
log.info("[REDIS: 推流设备状态变化] 更新部分设备离线： {}个", streamStatusMessage.getOfflineStreams().size());
streamPushService.offline(streamStatusMessage.getOfflineStreams());
streamPushService.offlineforRedisMsg(streamStatusMessage.getOfflineStreams());
}
if (streamStatusMessage.getOnlineStreams() != null &&
!streamStatusMessage.getOnlineStreams().isEmpty()) {
// 更新部分设备上线
log.info("[REDIS: 推流设备状态变化] 更新部分设备上线： {}个", streamStatusMessage.getOnlineStreams().size());
streamPushService.online(streamStatusMessage.getOnlineStreams());
streamPushService.onlineForRedisMsg(streamStatusMessage.getOnlineStreams());
}
} catch (Exception e) {
log.warn("[REDIS消息-推流设备状态变化] 发现未处理的异常, \r\n{}", JSON.parseObject(msg.getBody()));
@@ -115,7 +115,7 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
dynamicTask.startDelay(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED, () -> {
log.info("[REDIS消息]未收到redis回复推流设备状态，执行推流设备离线");
// 五秒收不到请求就设置通道离线，然后通知上级离线
streamPushService.allOffline();
streamPushService.allOfflineForRedisMsg();
}, 5000);
}

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

@@ -524,7 +524,8 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
if (serverId != null) {
redisTemplate.opsForZSet().remove(key, serverId);
}
Set<Object> range = redisTemplate.opsForZSet().range(key, 0, 0);
// 获取得分最高的，也是最后更新时间到redis的wvp，这样可以避免读取到离线的wvp，同时时间最新也一定程度代表最健康的
Set<Object> range = redisTemplate.opsForZSet().reverseRange(key, 0, 0);
if (range == null || range.isEmpty()) {
return null;
}

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

@@ -289,7 +289,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}
if (!channelListForOnline.isEmpty()) {
gbChannelService.online(channelListForOnline);
gbChannelService.online(channelListForOnline, true);
}
List<CommonGBChannel> channelListForOffline = new ArrayList<>();
List<StreamProxy> streamProxiesForRemove = new ArrayList<>();
@@ -302,7 +302,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}
}
if (!channelListForOffline.isEmpty()) {
gbChannelService.offline(channelListForOffline);
gbChannelService.offline(channelListForOffline, true);
}
if (!streamProxiesForRemove.isEmpty()) {
streamProxyMapper.deleteByList(streamProxiesForRemove);
@@ -338,7 +338,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}
if (!channelListForOffline.isEmpty()) {
// 修改国标关联的国标通道的状态
gbChannelService.offline(channelListForOffline);
gbChannelService.offline(channelListForOffline, true);
}
if (!streamProxiesForSendMessage.isEmpty()) {
for (StreamProxy streamProxy : streamProxiesForSendMessage) {

4
src/main/java/com/genersoft/iot/vmp/streamPush/dao/StreamPushMapper.java
View File

@@ -160,6 +160,6 @@ public interface StreamPushMapper {
@Delete(" DELETE FROM wvp_stream_push st " +
" LEFT join wvp_device_channel wdc on wdc.data_type = 2 and st.id = wdc.data_device_id " +
" where wdc.id is null and st.server_id = #{id}")
void deleteWithoutGBId(String id);
" where wdc.id is null and st.server_id = #{serverId}")
void deleteWithoutGBId(@Param("serverId") String serverId);
}

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

@@ -52,17 +52,17 @@ public interface IStreamPushService {
/**
* 全部离线
*/
void allOffline();
void allOfflineForRedisMsg();
/**
* 推流离线
*/
void offline(List<StreamPushItemFromRedis> offlineStreams);
void offlineforRedisMsg(List<StreamPushItemFromRedis> offlineStreams);
/**
* 推流上线
*/
void online(List<StreamPushItemFromRedis> onlineStreams);
void onlineForRedisMsg(List<StreamPushItemFromRedis> onlineStreams);
/**
* 增加推流
@@ -91,7 +91,7 @@ public interface IStreamPushService {
void updatePushStatus(StreamPush streamPush);
void batchUpdate(List<StreamPush> streamPushItemForUpdate);
void batchUpdateForRedisMsg(List<StreamPush> streamPushItemForUpdate);
int delete(int id);

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

@@ -446,7 +446,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
}
@Override
public void allOffline() {
public void allOfflineForRedisMsg() {
String serverId = redisCatchStorage.chooseOneServer(null);
boolean permission = userSetting.getServerId().equals(serverId);
List<StreamPush> streamPushList = streamPushMapper.selectAll(null, null, null);
if (streamPushList.isEmpty()) {
return;
@@ -458,11 +460,13 @@ public class StreamPushServiceImpl implements IStreamPushService {
commonGBChannelList.add(streamPush.buildCommonGBChannel());
}
}
gbChannelService.offline(commonGBChannelList);
gbChannelService.offline(commonGBChannelList, permission);
}
@Override
public void offline(List<StreamPushItemFromRedis> offlineStreams) {
public void offlineforRedisMsg(List<StreamPushItemFromRedis> offlineStreams) {
String serverId = redisCatchStorage.chooseOneServer(null);
boolean permission = userSetting.getServerId().equals(serverId);
// 更新部分设备离线
List<StreamPush> streamPushList = streamPushMapper.getListInList(offlineStreams);
if (streamPushList.isEmpty()) {
@@ -470,15 +474,17 @@ public class StreamPushServiceImpl implements IStreamPushService {
return;
}
List<CommonGBChannel> commonGBChannelList = gbChannelService.queryListByStreamPushList(streamPushList);
gbChannelService.offline(commonGBChannelList);
gbChannelService.offline(commonGBChannelList, permission);
}
@Override
public void online(List<StreamPushItemFromRedis> onlineStreams) {
public void onlineForRedisMsg(List<StreamPushItemFromRedis> onlineStreams) {
if (onlineStreams.isEmpty()) {
log.info("[设备上线] 推流设备列表为空");
return;
}
String serverId = redisCatchStorage.chooseOneServer(null);
boolean permission = userSetting.getServerId().equals(serverId);
// 更新部分设备上线streamPushService
List<StreamPush> streamPushList = streamPushMapper.getListInList(onlineStreams);
if (streamPushList.isEmpty()) {
@@ -488,7 +494,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
return;
}
List<CommonGBChannel> commonGBChannelList = gbChannelService.queryListByStreamPushList(streamPushList);
gbChannelService.online(commonGBChannelList);
gbChannelService.online(commonGBChannelList, permission);
}
@Override
@@ -556,15 +562,19 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
@Transactional
public void batchUpdate(List<StreamPush> streamPushItemForUpdate) {
streamPushMapper.batchUpdate(streamPushItemForUpdate);
public void batchUpdateForRedisMsg(List<StreamPush> streamPushItemForUpdate) {
String serverId = redisCatchStorage.chooseOneServer(null);
boolean permission = userSetting.getServerId().equals(serverId);
if (permission) {
streamPushMapper.batchUpdate(streamPushItemForUpdate);
}
List<CommonGBChannel> commonGBChannels = new ArrayList<>();
for (StreamPush streamPush : streamPushItemForUpdate) {
if (!ObjectUtils.isEmpty(streamPush.getGbDeviceId())) {
commonGBChannels.add(streamPush.buildCommonGBChannel());
}
}
gbChannelService.batchUpdate(commonGBChannels);
gbChannelService.batchUpdateForStreamPushRedisMsg(commonGBChannels, permission);
}
@Override

1
src/main/java/com/genersoft/iot/vmp/web/custom/bean/SYMember.java
View File

@@ -12,5 +12,6 @@ public class SYMember {
private Long blockId;
private String unitNo;
private String terminalMemberStatus;
private String channelDeviceId;
}

72
src/main/java/com/genersoft/iot/vmp/web/custom/service/CameraChannelService.java
View File

@@ -42,6 +42,7 @@ public class CameraChannelService implements CommandLineRunner {
private final String REDIS_CHANNEL_MESSAGE = "VM_MSG_MOBILE_CHANNEL";
private final String REDIS_MEMBER_STATUS_MESSAGE = "VM_MSG_MEMBER_STATUS_CHANNEL";
private final String MOBILE_CHANNEL_PREFIX = "nationalStandardMobileTerminal_";
private final String DELAY_TASK_KEY = "DELAY_TASK_KEY_";
@Autowired
private CommonGBChannelMapper channelMapper;
@@ -126,9 +127,9 @@ public class CameraChannelService implements CommandLineRunner {
List<CommonGBChannel> resultListForOnline = new ArrayList<>();
List<CommonGBChannel> resultListForOffline = new ArrayList<>();
List<SYMember> memberList = new ArrayList<>();
List<CommonGBChannel> addMemberList = new ArrayList<>();
Map<String, CommonGBChannel> delayChannelMap = new HashMap<>();
List<SYMember> memberList = new ArrayList<>();
switch (event.getMessageType()) {
@@ -157,16 +158,29 @@ public class CameraChannelService implements CommandLineRunner {
if (oldChannel != null) {
if (oldChannel.getGbPtzType() != null && oldChannel.getGbPtzType() == 99) {
resultListForUpdate.add(channel);
// 如果状态变化发送消息
if (!Objects.equals(oldChannel.getGbStatus(), channel.getGbStatus())) {
SYMember member = getMember(channel.getGbDeviceId());
if (member != null) {
if ("ON".equals(channel.getGbStatus())) {
member.setTerminalMemberStatus("ONLINE");
}else {
member.setTerminalMemberStatus("OFFLINE");
}
memberList.add(member);
}
}
}else {
resultListForAdd.add(channel);
if ("ON".equals(channel.getGbStatus())) {
addMemberList.add(channel);
delayChannelMap.put(channel.getGbDeviceId(), channel);
}
}
}else {
resultListForAdd.add(channel);
if ("ON".equals(channel.getGbStatus())) {
addMemberList.add(channel);
delayChannelMap.put(channel.getGbDeviceId(), channel);
}
}
}else {
@@ -175,6 +189,11 @@ public class CameraChannelService implements CommandLineRunner {
CameraChannel cameraChannel = new CameraChannel();
cameraChannel.setGbDeviceId(channel.getGbDeviceId());
resultListForDelete.add(cameraChannel);
SYMember member = getMember(cameraChannel.getGbDeviceId());
if (member != null) {
member.setTerminalMemberStatus("OFFLINE");
memberList.add(member);
}
}
}
}
@@ -186,8 +205,14 @@ public class CameraChannelService implements CommandLineRunner {
CameraChannel cameraChannel = new CameraChannel();
cameraChannel.setGbDeviceId(channel.getGbDeviceId());
resultListForDelete.add(cameraChannel);
SYMember member = getMember(cameraChannel.getGbDeviceId());
if (member != null) {
member.setTerminalMemberStatus("OFFLINE");
memberList.add(member);
}
}
}
break;
case ON:
case OFF:
@@ -221,7 +246,7 @@ public class CameraChannelService implements CommandLineRunner {
if (channel.getGbPtzType() != null && channel.getGbPtzType() == 99) {
resultListForAdd.add(channel);
if ("ON".equals(channel.getGbStatus())) {
addMemberList.add(channel);
delayChannelMap.put(channel.getGbDeviceId(), channel);
}
}
}
@@ -249,27 +274,34 @@ public class CameraChannelService implements CommandLineRunner {
if (!memberList.isEmpty()) {
sendMemberStatusMessage(memberList);
}
if (!addMemberList.isEmpty()) {
if (!delayChannelMap.isEmpty()) {
// 对于在线的终端进行延迟检查和发送
String key = UUID.randomUUID().toString();
dynamicTask.startDelay(key, () -> {
List<SYMember> members = new ArrayList<>();
for (CommonGBChannel commonGBChannel : addMemberList) {
for (CommonGBChannel commonGBChannel : delayChannelMap.values()) {
String key = DELAY_TASK_KEY + commonGBChannel.getGbDeviceId();
dynamicTask.startDelay(key, () -> {
dynamicTask.stop(key);
SYMember member = getMember(commonGBChannel.getGbDeviceId());
if (member == null) {
continue;
return;
}
member.setTerminalMemberStatus("ONLINE");
members.add(member);
}
if (!members.isEmpty()) {
sendMemberStatusMessage(members);
}
}, 5000);
sendMemberStatusMessage(List.of(member));
}, 3000);
}
}
}
private void sendMemberStatusMessage(List<SYMember> memberList) {
// 取消延时发送
for (SYMember syMember : memberList) {
String key = DELAY_TASK_KEY + syMember.getChannelDeviceId();
if (dynamicTask.contains(key)) {
log.info("[SY-redis发送通知] 取消延时新增任务: {}", key);
dynamicTask.stop(key);
}
}
String jsonString = JSONObject.toJSONString(memberList);
log.info("[SY-redis发送通知] 发送 状态变化 {}: {}", REDIS_MEMBER_STATUS_MESSAGE, jsonString);
redisTemplateForString.convertAndSend(REDIS_MEMBER_STATUS_MESSAGE, jsonString);
@@ -277,6 +309,7 @@ public class CameraChannelService implements CommandLineRunner {
private void sendChannelMessage(List<CommonGBChannel> channelList, ChannelEvent.ChannelEventMessageType type) {
if (channelList.isEmpty()) {
log.warn("[SY-redis发送通知-{}] 发送失败，数据为空， 通道信息变化 {}", type, REDIS_CHANNEL_MESSAGE);
return;
}
List<CameraChannel> cameraChannelList = channelMapper.queryCameraChannelByIds(channelList);
@@ -311,6 +344,7 @@ public class CameraChannelService implements CommandLineRunner {
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
jsonObject.put("blockId", member.getBlockId());
jsonObject.put("gbDeviceId", mobilePosition.getChannelDeviceId());
log.info("[SY-redis发送通知-移动设备位置信息] 发送 {}: {}", REDIS_GPS_MESSAGE, jsonObject.toString());
redisTemplateForString.convertAndSend(REDIS_GPS_MESSAGE, jsonObject.toString());
}
@@ -322,7 +356,9 @@ public class CameraChannelService implements CommandLineRunner {
if (jsonObject == null) {
return null;
}
return JSONObject.parseObject(jsonObject.toString(), SYMember.class);
SYMember syMember = JSONObject.parseObject(jsonObject.toString(), SYMember.class);
syMember.setChannelDeviceId(deviceId);
return syMember;
}

22
web/src/views/common/CommonChannelEdit.vue
View File

@@ -30,11 +30,14 @@
</el-form-item>
<el-form-item label="行政区域">
<el-input v-model="form.gbCivilCode" placeholder="请输入行政区域">
<el-input v-model="form.gbCivilCode" placeholder="请输入行政区域" @change="getRegionPaths">
<template v-slot:append>
<el-button @click="chooseCivilCode()">选择</el-button>
</template>
</el-input>
<el-breadcrumb v-if="regionPath.length > 0" separator="/" style="display: block; margin-top: 8px; font-size: 14px;">
<el-breadcrumb-item v-for="key in regionPath" :key="key">{{ key }}</el-breadcrumb-item>
</el-breadcrumb>
</el-form-item>
<el-form-item label="安装地址">
@@ -274,6 +277,7 @@ export default {
loading: false,
modelList: [],
parentPath: [],
regionPath: [],
form: {}
}
},
@@ -387,6 +391,7 @@ export default {
this.form = data
this.$set(this.form, 'enableBroadcastForBool', this.form.enableBroadcast === 1)
this.getPaths()
this.getRegionPaths()
})
.finally(() => {
this.loading = false
@@ -400,6 +405,7 @@ export default {
chooseCivilCode: function() {
this.$refs.chooseCivilCode.openDialog(code => {
this.form.gbCivilCode = code
this.getRegionPaths()
})
},
chooseGroup: function() {
@@ -431,6 +437,20 @@ export default {
this.parentPath = path
})
}
},
getRegionPaths: function() {
this.regionPath = []
if (this.form.gbCivilCode) {
this.$store.dispatch('region/queryPath', this.form.gbCivilCode)
.then(data => {
console.log(data)
const path = []
for (let i = 0; i < data.length; i++) {
path.push(data[i].name)
}
this.regionPath = path
})
}
}
}
}