Merge branch 'talk' into main-dev

648540858

2023-03-20 039fbf7e243d4bf0f46b44e7dae2a5d36e978056

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

@@ -6,7 +6,6 @@
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -90,12 +89,6 @@
   @Autowired
    private PlatformGbStreamMapper platformGbStreamMapper;

   @Autowired
    private IGbStreamService gbStreamService;

   @Autowired
    private ParentPlatformMapper parentPlatformMapper;

   /**
    * 根据设备ID判断设备是否存在
    *
@@ -133,6 +126,15 @@
               if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
                  deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
                  deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
                  if (allChannelMap.get(deviceChannel.getChannelId()).getStatus() !=deviceChannel.getStatus()){
                     List<String> strings = platformChannelMapper.queryParentPlatformByChannelId(deviceChannel.getChannelId());
                     if (!CollectionUtils.isEmpty(strings)){
                        strings.forEach(platformId->{
                           eventPublisher.catalogEventPublish(platformId, deviceChannel, deviceChannel.getStatus()==1?CatalogEvent.ON:CatalogEvent.OFF);
                        });
                     }

                  }
               }
               channels.add(deviceChannel);
               if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
@@ -192,6 +194,121 @@
         return false;
      }

   }


   @Override
   public boolean updateChannels(String deviceId, List<DeviceChannel> deviceChannelList) {
      if (CollectionUtils.isEmpty(deviceChannelList)) {
         return false;
      }
      List<DeviceChannel> allChannels = deviceChannelMapper.queryAllChannels(deviceId);
      Map<String,DeviceChannel> allChannelMap = new ConcurrentHashMap<>();
      if (allChannels.size() > 0) {
         for (DeviceChannel deviceChannel : allChannels) {
            allChannelMap.put(deviceChannel.getChannelId(), deviceChannel);
         }
      }
      List<DeviceChannel> addChannels = new ArrayList<>();
      List<DeviceChannel> updateChannels = new ArrayList<>();


      TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
      // 数据去重
      StringBuilder stringBuilder = new StringBuilder();
      Map<String, Integer> subContMap = new HashMap<>();
      if (deviceChannelList.size() > 0) {
         // 数据去重
         Set<String> gbIdSet = new HashSet<>();
         for (DeviceChannel deviceChannel : deviceChannelList) {
            if (!gbIdSet.contains(deviceChannel.getChannelId())) {
               gbIdSet.add(deviceChannel.getChannelId());
               if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
                  deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
                  deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
                  deviceChannel.setUpdateTime(DateUtil.getNow());
                  updateChannels.add(deviceChannel);
               }else {
                  deviceChannel.setCreateTime(DateUtil.getNow());
                  addChannels.add(deviceChannel);
               }
               if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
                  if (subContMap.get(deviceChannel.getParentId()) == null) {
                     subContMap.put(deviceChannel.getParentId(), 1);
                  }else {
                     Integer count = subContMap.get(deviceChannel.getParentId());
                     subContMap.put(deviceChannel.getParentId(), count++);
                  }
               }
            }else {
               stringBuilder.append(deviceChannel.getChannelId()).append(",");
            }
         }
         if (addChannels.size() > 0) {
            for (DeviceChannel channel : addChannels) {
               if (subContMap.get(channel.getChannelId()) != null){
                  channel.setSubCount(subContMap.get(channel.getChannelId()));
               }
            }
         }
         if (updateChannels.size() > 0) {
            for (DeviceChannel channel : updateChannels) {
               if (subContMap.get(channel.getChannelId()) != null){
                  channel.setSubCount(subContMap.get(channel.getChannelId()));
               }
            }
         }

      }
      if (stringBuilder.length() > 0) {
         logger.info("[目录查询]收到的数据存在重复： {}" , stringBuilder);
      }
      if(CollectionUtils.isEmpty(updateChannels) && CollectionUtils.isEmpty(addChannels) ){
         logger.info("通道更新，数据为空={}" , deviceChannelList);
         return false;
      }
      try {
         int limitCount = 300;
         boolean result = false;
         if (addChannels.size() > 0) {
            if (addChannels.size() > limitCount) {
               for (int i = 0; i < addChannels.size(); i += limitCount) {
                  int toIndex = i + limitCount;
                  if (i + limitCount > addChannels.size()) {
                     toIndex = addChannels.size();
                  }
                  result = result || deviceChannelMapper.batchAdd(addChannels.subList(i, toIndex)) < 0;
               }
            }else {
               result = result || deviceChannelMapper.batchAdd(addChannels) < 0;
            }
         }
         if (updateChannels.size() > 0) {
            if (updateChannels.size() > limitCount) {
               for (int i = 0; i < updateChannels.size(); i += limitCount) {
                  int toIndex = i + limitCount;
                  if (i + limitCount > updateChannels.size()) {
                     toIndex = updateChannels.size();
                  }
                  result = result || deviceChannelMapper.batchUpdate(updateChannels.subList(i, toIndex)) < 0;
               }
            }else {
               result = result || deviceChannelMapper.batchUpdate(updateChannels) < 0;
            }
         }
         if (result) {
            //事务回滚
            dataSourceTransactionManager.rollback(transactionStatus);
         }else {
            //手动提交
            dataSourceTransactionManager.commit(transactionStatus);
         }
         return true;
      }catch (Exception e) {
         e.printStackTrace();
         dataSourceTransactionManager.rollback(transactionStatus);
         return false;
      }
   }

   @Override
@@ -471,6 +588,20 @@

   }

   @Override
   public Device queryDeviceInfoByPlatformIdAndChannelId(String platformId, String channelId) {
      List<Device> devices = platformChannelMapper.queryDeviceInfoByPlatformIdAndChannelId(platformId, channelId);
      if (devices.size() > 1) {
         // 出现长度大于0的时候肯定是国标通道的ID重复了
         logger.warn("国标ID存在重复：{}", channelId);
      }
      if (devices.size() == 0) {
         return null;
      }else {
         return devices.get(0);
      }
   }

   /**
    * 查询最新移动位置
    * @param deviceId