From a1671c3c3ee5114d5fd5bd032f8245f8a22c010f Mon Sep 17 00:00:00 2001 From: panlinlin <648540858@qq.com> Date: Thu, 8 Aug 2024 23:44:03 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=B4=E6=97=B6=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/gb28181/bean/CatalogData.java | 62 ++----------------- .../genersoft/iot/vmp/gb28181/bean/Group.java | 20 ++++++ .../iot/vmp/gb28181/bean/Region.java | 14 +++++ .../iot/vmp/gb28181/dao/GroupMapper.java | 8 +++ .../iot/vmp/gb28181/dao/RegionMapper.java | 7 +++ .../vmp/gb28181/service/IGroupService.java | 2 + .../vmp/gb28181/service/IRegionService.java | 2 + .../service/impl/GroupServiceImpl.java | 22 +++++++ .../service/impl/RegionServiceImpl.java | 23 +++++++ .../vmp/gb28181/session/CatalogDataCatch.java | 30 +++++++-- .../cmd/CatalogResponseMessageHandler.java | 56 +++++++++++++---- 11 files changed, 172 insertions(+), 74 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java index 965d7f2a..4666b707 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java @@ -1,11 +1,14 @@ package com.genersoft.iot.vmp.gb28181.bean; +import lombok.Data; + import java.time.Instant; import java.util.List; /** * @author lin */ +@Data public class CatalogData { /** * 命令序列号 @@ -13,6 +16,8 @@ public class CatalogData { private int sn; private int total; private List channelList; + private List regionListList; + private List groupListListList; private Instant lastTime; private Device device; private String errorMsg; @@ -21,61 +26,4 @@ public class CatalogData { ready, runIng, end } private CatalogDataStatus status; - - - public int getSn() { - return sn; - } - - public void setSn(int sn) { - this.sn = sn; - } - - public int getTotal() { - return total; - } - - public void setTotal(int total) { - this.total = total; - } - - public List getChannelList() { - return channelList; - } - - public void setChannelList(List channelList) { - this.channelList = channelList; - } - - public Instant getLastTime() { - return lastTime; - } - - public void setLastTime(Instant lastTime) { - this.lastTime = lastTime; - } - - public Device getDevice() { - return device; - } - - public void setDevice(Device device) { - this.device = device; - } - - public String getErrorMsg() { - return errorMsg; - } - - public void setErrorMsg(String errorMsg) { - this.errorMsg = errorMsg; - } - - public CatalogDataStatus getStatus() { - return status; - } - - public void setStatus(CatalogDataStatus status) { - this.status = status; - } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Group.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Group.java index d7f1a142..e636f003 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Group.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Group.java @@ -1,5 +1,6 @@ package com.genersoft.iot.vmp.gb28181.bean; +import com.genersoft.iot.vmp.utils.DateUtil; import io.swagger.v3.oas.annotations.media.Schema; import lombok.Data; import org.jetbrains.annotations.NotNull; @@ -58,6 +59,25 @@ public class Group implements Comparable{ @Schema(description = "平台ID") private Integer platformId; + public static Group getInstance(DeviceChannel channel) { + GbCode gbCode = GbCode.decode(channel.getDeviceId()); + if (gbCode == null || (!gbCode.getTypeCode().equals("215") && !gbCode.getTypeCode().equals("216"))) { + return null; + } + Group group = new Group(); + group.setName(channel.getName()); + group.setDeviceId(channel.getDeviceId()); + group.setCreateTime(DateUtil.getNow()); + group.setUpdateTime(DateUtil.getNow()); + if (gbCode.getTypeCode().equals("215")) { + group.setBusinessGroup(channel.getDeviceId()); + }else if (gbCode.getTypeCode().equals("216")) { + group.setBusinessGroup(channel.getBusinessGroupId()); + group.setParentDeviceId(channel.getParentId()); + } + return group; + } + @Override public int compareTo(@NotNull Group region) { return Integer.compare(Integer.parseInt(this.deviceId), Integer.parseInt(region.getDeviceId())); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Region.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Region.java index cabba0af..e1e99f0c 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Region.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Region.java @@ -1,6 +1,7 @@ package com.genersoft.iot.vmp.gb28181.bean; import com.genersoft.iot.vmp.common.CivilCodePo; +import com.genersoft.iot.vmp.utils.CivilCodeUtil; import com.genersoft.iot.vmp.utils.DateUtil; import io.swagger.v3.oas.annotations.media.Schema; import lombok.Data; @@ -70,6 +71,19 @@ public class Region implements Comparable{ return region; } + public static Region getInstance(DeviceChannel channel) { + Region region = new Region(); + region.setName(channel.getName()); + region.setDeviceId(channel.getDeviceId()); + CivilCodePo parentCode = CivilCodeUtil.INSTANCE.getParentCode(channel.getDeviceId()); + if (parentCode != null) { + region.setParentDeviceId(parentCode.getCode()); + } + region.setCreateTime(DateUtil.getNow()); + region.setUpdateTime(DateUtil.getNow()); + return region; + } + @Override public int compareTo(@NotNull Region region) { return Integer.compare(Integer.parseInt(this.deviceId), Integer.parseInt(region.getDeviceId())); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/dao/GroupMapper.java b/src/main/java/com/genersoft/iot/vmp/gb28181/dao/GroupMapper.java index 5dfd4228..212e722b 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/dao/GroupMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/dao/GroupMapper.java @@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.dao; import com.genersoft.iot.vmp.gb28181.bean.Group; import com.genersoft.iot.vmp.gb28181.bean.GroupTree; +import com.genersoft.iot.vmp.gb28181.bean.Region; import org.apache.ibatis.annotations.*; import java.util.List; @@ -152,4 +153,11 @@ public interface GroupMapper { " SET parent_device_id=#{group.deviceId}, business_group = #{group.businessGroup}" + " WHERE parent_device_id = #{oldDeviceId}") int updateChild(@Param("oldDeviceId") String oldDeviceId, Group group); + + @Select(" ") + List queryInGroupList(List groupList); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/dao/RegionMapper.java b/src/main/java/com/genersoft/iot/vmp/gb28181/dao/RegionMapper.java index 6521bac0..2289ffb2 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/dao/RegionMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/dao/RegionMapper.java @@ -89,4 +89,11 @@ public interface RegionMapper { " #{item.id}" + " ") void batchDelete(List allChildren); + + @Select(" ") + List queryInRegionList(List regionList); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/IGroupService.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/IGroupService.java index 8be65616..67629645 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/IGroupService.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/IGroupService.java @@ -19,4 +19,6 @@ public interface IGroupService { void syncFromChannel(); boolean delete(int id); + + boolean batchAdd(List groupList); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/IRegionService.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/IRegionService.java index 19c841da..a1f7477a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/IRegionService.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/IRegionService.java @@ -37,4 +37,6 @@ public interface IRegionService { void syncFromChannel(); boolean delete(int id); + + boolean batchAdd(List regionList); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GroupServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GroupServiceImpl.java index 108fb833..ea5f514d 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GroupServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GroupServiceImpl.java @@ -16,7 +16,9 @@ import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; /** * 区域管理类 @@ -234,4 +236,24 @@ public class GroupServiceImpl implements IGroupService { } return true; } + + @Override + public boolean batchAdd(List groupList) { + if (groupList== null || groupList.isEmpty()) { + return false; + } + Map groupMapForVerification = new HashMap<>(); + for (Group group : groupList) { + groupMapForVerification.put(group.getDeviceId(), group); + } + // 查询数据库中已经存在的. + List regionListInDb = groupManager.queryInGroupList(groupList); + if (!regionListInDb.isEmpty()) { + for (Region region : regionListInDb) { + groupMapForVerification.remove(region.getDeviceId()); + } + } + groupManager.batchAdd(new ArrayList<>(groupMapForVerification.values())); + return false; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/RegionServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/RegionServiceImpl.java index b34830bb..9824beba 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/RegionServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/RegionServiceImpl.java @@ -13,6 +13,8 @@ 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.dao.DuplicateKeyException; import org.springframework.stereotype.Service; @@ -29,6 +31,7 @@ import java.util.*; public class RegionServiceImpl implements IRegionService { + private static final Logger log = LoggerFactory.getLogger(RegionServiceImpl.class); @Autowired private RegionMapper regionMapper; @@ -177,4 +180,24 @@ public class RegionServiceImpl implements IRegionService { public boolean delete(int id) { return regionMapper.delete(id) > 0; } + + @Override + public boolean batchAdd(List regionList) { + if (regionList== null || regionList.isEmpty()) { + return false; + } + Map regionMapForVerification = new HashMap<>(); + for (Region region : regionList) { + regionMapForVerification.put(region.getDeviceId(), region); + } + // 查询数据库中已经存在的. + List regionListInDb = regionMapper.queryInRegionList(regionList); + if (!regionListInDb.isEmpty()) { + for (Region region : regionListInDb) { + regionMapForVerification.remove(region.getDeviceId()); + } + } + regionMapper.batchAdd(new ArrayList<>(regionMapForVerification.values())); + return false; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java index 6642c854..b0ac22a9 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java @@ -1,9 +1,6 @@ package com.genersoft.iot.vmp.gb28181.session; -import com.genersoft.iot.vmp.gb28181.bean.CatalogData; -import com.genersoft.iot.vmp.gb28181.bean.Device; -import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; -import com.genersoft.iot.vmp.gb28181.bean.SyncStatus; +import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.service.IDeviceChannelService; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import org.springframework.beans.factory.annotation.Autowired; @@ -39,7 +36,8 @@ public class CatalogDataCatch { } } - public void put(String deviceId, int sn, int total, Device device, List deviceChannelList) { + public void put(String deviceId, int sn, int total, Device device, List deviceChannelList, + List regionList, List groupList) { CatalogData catalogData = data.get(deviceId); if (catalogData == null) { catalogData = new CatalogData(); @@ -47,6 +45,8 @@ public class CatalogDataCatch { catalogData.setTotal(total); catalogData.setDevice(device); catalogData.setChannelList(deviceChannelList); + catalogData.setRegionListList(regionList); + catalogData.setGroupListListList(groupList); catalogData.setStatus(CatalogData.CatalogDataStatus.runIng); catalogData.setLastTime(Instant.now()); data.put(deviceId, catalogData); @@ -59,11 +59,13 @@ public class CatalogDataCatch { catalogData.setDevice(device); catalogData.setStatus(CatalogData.CatalogDataStatus.runIng); catalogData.getChannelList().addAll(deviceChannelList); + catalogData.getRegionListList().addAll(regionList); + catalogData.getGroupListListList().addAll(groupList); catalogData.setLastTime(Instant.now()); } } - public List get(String deviceId) { + public List getDeviceChannelList(String deviceId) { CatalogData catalogData = data.get(deviceId); if (catalogData == null) { return null; @@ -71,6 +73,22 @@ public class CatalogDataCatch { return catalogData.getChannelList(); } + public List getRegionList(String deviceId) { + CatalogData catalogData = data.get(deviceId); + if (catalogData == null) { + return null; + } + return catalogData.getRegionListList(); + } + + public List getGroupList(String deviceId) { + CatalogData catalogData = data.get(deviceId); + if (catalogData == null) { + return null; + } + return catalogData.getGroupListListList(); + } + public int getTotal(String deviceId) { CatalogData catalogData = data.get(deviceId); if (catalogData == null) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java index a2f44d45..c4a663b6 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java @@ -2,12 +2,13 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.respon import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.gb28181.bean.*; +import com.genersoft.iot.vmp.gb28181.service.IGroupService; +import com.genersoft.iot.vmp.gb28181.service.IRegionService; import com.genersoft.iot.vmp.gb28181.session.CatalogDataCatch; import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler; import com.genersoft.iot.vmp.gb28181.service.IDeviceChannelService; -import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import gov.nist.javax.sip.message.SIPRequest; import lombok.extern.slf4j.Slf4j; import org.dom4j.DocumentException; @@ -17,6 +18,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; import javax.sip.InvalidArgumentException; import javax.sip.RequestEvent; @@ -44,10 +46,13 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); @Autowired - private IVideoManagerStorage storager; + private IDeviceChannelService deviceChannelService; @Autowired - private IDeviceChannelService deviceChannelService; + private IRegionService regionService; + + @Autowired + private IGroupService groupService; @Autowired private CatalogDataCatch catalogDataCatch; @@ -66,6 +71,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp } @Override + @Transactional public void handForDevice(RequestEvent evt, Device device, Element element) { taskQueue.offer(new HandlerCatchData(evt, device, element)); // 回复200 OK @@ -106,6 +112,8 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp Iterator deviceListIterator = deviceListElement.elementIterator(); if (deviceListIterator != null) { List channelList = new ArrayList<>(); + List regionList = new ArrayList<>(); + List groupList = new ArrayList<>(); // 遍历DeviceList while (deviceListIterator.hasNext()) { Element itemDevice = deviceListIterator.next(); @@ -114,7 +122,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp continue; } DeviceChannel channel = DeviceChannel.decode(itemDevice); - if (channel == null || channel.getDeviceId() == null) { + if (channel.getDeviceId() == null) { log.info("[收到目录订阅]：但是解析失败 {}", new String(evt.getRequest().getRawContent())); continue; } @@ -122,17 +130,30 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) { channel.setParentId(null); } + // 解析通道类型 + if (channel.getDeviceId().length() <= 8) { + // 行政区划 + Region region = Region.getInstance(channel); + regionList.add(region); + }else if (channel.getDeviceId().length() == 20){ + // 业务分组/虚拟组织 + Group group = Group.getInstance(channel); + if (group != null) { + groupList.add(group); + } + } channelList.add(channel); } int sn = Integer.parseInt(snElement.getText()); - catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), channelList); - log.info("[收到通道]设备: {} -> {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 : catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum); - if (catalogDataCatch.get(take.getDevice().getDeviceId()).size() == sumNum) { + catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), + channelList, regionList, groupList); + log.info("[收到通道]设备: {} -> {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.getDeviceChannelList(take.getDevice().getDeviceId()) == null ? 0 : catalogDataCatch.getDeviceChannelList(take.getDevice().getDeviceId()).size(), sumNum); + if (catalogDataCatch.getDeviceChannelList(take.getDevice().getDeviceId()).size() == sumNum) { // 数据已经完整接收， 此时可能存在某个设备离线变上线的情况，但是考虑到性能，此处不做处理， // 目前支持设备通道上线通知时和设备上线时向上级通知 - boolean resetChannelsResult = deviceChannelService.resetChannels(device.getId(), catalogDataCatch.get(take.getDevice().getDeviceId())); + boolean resetChannelsResult = saveData(device); if (!resetChannelsResult) { - String errorMsg = "接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(take.getDevice().getDeviceId()).size() + "条"; + String errorMsg = "接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.getDeviceChannelList(take.getDevice().getDeviceId()).size() + "条"; catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), errorMsg); } else { catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null); @@ -152,13 +173,26 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp } + @Transactional + public boolean saveData(Device device) { + + boolean result = deviceChannelService.resetChannels(device.getId(), catalogDataCatch.getDeviceChannelList(device.getDeviceId())); + if (!catalogDataCatch.getRegionList(device.getDeviceId()).isEmpty()) { + result &= regionService.batchAdd(catalogDataCatch.getRegionList(device.getDeviceId())); + } + if (!catalogDataCatch.getGroupList(device.getDeviceId()).isEmpty()) { + result &= groupService.batchAdd(catalogDataCatch.getGroupList(device.getDeviceId())); + } + return result; + } + @Override public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element rootElement) { } public SyncStatus getChannelSyncProgress(String deviceId) { - if (catalogDataCatch.get(deviceId) == null) { + if (catalogDataCatch.getDeviceChannelList(deviceId) == null) { return null; } else { return catalogDataCatch.getSyncStatus(deviceId); @@ -166,7 +200,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp } public boolean isSyncRunning(String deviceId) { - if (catalogDataCatch.get(deviceId) == null) { + if (catalogDataCatch.getDeviceChannelList(deviceId) == null) { return false; } else { return catalogDataCatch.isSyncRunning(deviceId);