北向传感器代码修改提交

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/controller/DataController.java

@@ -110,7 +110,7 @@ public class DataController {
 
 
 
-    @ApiOperation(value = "传感器通用设备信息同步", notes = "直连主机传感器")
+    @ApiOperation(value = "传感器通用设备信息同步", notes = "直连主机传感器,每个完整的包为一个机构机构数据,非所有机构数据")
     @PostMapping({"/api/data/sensorCommonList"})
     public ReceiveErrorDto sensor(String msgId, @RequestBody DataPageDto<SensorDto> sensors) {
         LogUtils.DIRECT_HOST_SENSOR.info("【北向接口-{}】【msgId:{}】【接收参数：{}】", "接受传感器通用设备信息同步:/api/data/sensorCommonList", msgId, JSON.toJSONString(sensors));

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/mapper/IotSensorMapper.java

@@ -4,6 +4,7 @@ package com.xunmei.mediator.api.mapper;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.xunmei.common.core.domain.iot.domain.IotSensor;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
 
 /**
  * <p>
@@ -16,4 +17,5 @@ import org.apache.ibatis.annotations.Mapper;
 @Mapper
 public interface IotSensorMapper extends BaseMapper<IotSensor> {
 
+    void updateDelByOrgId(@Param("orgId") Long orgId);
 }

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/service/IotSensorService.java

@@ -21,4 +21,6 @@ public interface IotSensorService extends IService<IotSensor> {
 
     ReceiveErrorDto saveDirectHostStatus(DataPageDto<SensorStatusDto> statuses, String msgId);
      void saveStatusAsync(DataPageDto<SensorStatusDto> request, String msgId);
+
+    void updateDelByOrgId(Long orgId);
 }

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/service/impl/IotSensorServiceImpl.java

@@ -27,6 +27,7 @@ import com.xunmei.system.api.dto.SensorStatusDto;
 import com.xunmei.system.api.dto.protection.ReceiveErrorDto;
 import com.xunmei.system.api.enums.ElectricityMeterAttributes;
 import com.xunmei.system.api.util.LogUtils;
+import io.netty.util.internal.StringUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
@@ -67,6 +68,15 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
 
     @Override
     @Transactional(rollbackFor = Exception.class)
+    public void updateDelByOrgId(Long orgId) {
+        if (ObjectUtil.isNull(orgId)) {
+            return;
+        }
+        baseMapper.updateDelByOrgId(orgId);
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
     public ReceiveErrorDto saveDirectHostStatus(DataPageDto<SensorStatusDto> request, String msgId) {
         ReceiveErrorDto dto = CheckDataUtil.checkObjFieldIsNull(request);
 
@@ -78,11 +88,24 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
     }
 
     @Override
-    public void saveStatusAsync(DataPageDto<SensorStatusDto> request, String msgId){
+    public void saveStatusAsync(DataPageDto<SensorStatusDto> request, String msgId) {
         String path = "/api/status/sensorCommon";
         List<NorthError> errors = new ArrayList<>();
-        Map<String, List<IotSensor>> productDeviceMap = getAllDevice();
-        if (ObjectUtil.isEmpty(productDeviceMap)){
+        final long count = request.getData().stream().map(SensorStatusDto::getOrgCode).distinct().count();
+        if (count > 1) {
+            errors.add(new NorthError(msgId, path, StringUtil.EMPTY_STRING, "数据异常：完整数据内发现存在不同机构数据"));
+            northErrorService.saveErrorData(errors);
+            return;
+        }
+        final SensorStatusDto dto = request.getData().get(0);
+        final SysOrg org = remoteOrgService.findByCode(dto.getOrgCode(), SecurityConstants.INNER);
+        if (org == null) {
+            errors.add(new NorthError(msgId, path, dto, "参数非法：机构orgCode不存在:" + dto.getOrgCode()));
+            northErrorService.saveErrorData(errors);
+            return;
+        }
+        Map<String, List<IotSensor>> productDeviceMap = getAllDevice(org.getId());
+        if (ObjectUtil.isEmpty(productDeviceMap)) {
             LogUtils.DIRECT_HOST_SENSOR_STATUS.error("处理动环传感器状态时,未查询到传感器数据!");
             return;
         }
@@ -94,17 +117,14 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
                         , status, "参数非法：未传orgCode"));
                 continue;
             }
-            final SysOrg org = remoteOrgService.findByCode(status.getOrgCode(), SecurityConstants.INNER);
-            if (org == null) {
-                errors.add(new NorthError(msgId, path
-                        , status, "参数非法：机构orgCode不存在:" + status.getOrgCode()));
+            final List<IotSensor> iotSensors = productDeviceMap.get(String.valueOf(status.getDeviceCode()));
+            if (ObjectUtil.isEmpty(iotSensors)) {
+                errors.add(new NorthError(msgId, path, status, "对应设备不存在:" + status.getDeviceCode()));
                 continue;
             }
-            final List<IotSensor> iotSensors = productDeviceMap.get(String.valueOf(org.getId()));
             final Optional<IotSensor> optional = iotSensors.stream().filter(sensor -> ObjectUtil.equal(sensor.getDeviceCode(), status.getDeviceCode())).findFirst();
             if (!optional.isPresent()) {
-                errors.add(new NorthError(msgId, path
-                        , status, "对应设备不存在:" + status.getDeviceCode()));
+                errors.add(new NorthError(msgId, path, status, "对应设备不存在:" + status.getDeviceCode()));
                 continue;
             }
             final String infoStr = JSON.toJSONString(status.getInfos());
@@ -173,35 +193,43 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
     public ReceiveErrorDto saveDirectHost(DataPageDto<SensorDto> request, String msgId) {
         //验证基础分页信息
         ReceiveErrorDto dto = CheckDataUtil.checkObjFieldIsNull(request);
-
-
         if (dto.getSuccess()) {
             //分页数据的存取
             final List<SensorDto> data = RedisCheckRepeatDataUtil.isCompleted(request, SensorDto.class);
-
             LogUtils.DIRECT_HOST_SENSOR.info("判断是否获取通用传感器全部信息分页数据:{}", data.size() > 0);
             if (ObjectUtil.isEmpty(data)) {
                 return dto;
             }
+            List<IotSensor> devices = convertToDomain(data, msgId);
+            if (ObjectUtil.isEmpty(devices)) {
 
-           /* new Thread(() -> {
-                //删除直连主机传感器一个月前的状态数据
-                // productDeviceMapper.deleteDirectHostHistory(DateUtil.offset(new Date(), DateField.MONTH, -1));
-            }).start();*/
-
-            List<Long> deletedDeviceIds = new ArrayList<>();
-            List<IotSensor> devices = convertToDomain(data, msgId, deletedDeviceIds);
-            this.removeByIds(deletedDeviceIds);
+                return dto;
+            }
+            final Long orgId = devices.get(0).getOrgId();
+            this.updateDelByOrgId(orgId);
             this.saveOrUpdateBatch(devices);
         }
         return dto;
     }
 
-    private List<IotSensor> convertToDomain(List<SensorDto> sensorDtos, String msgId, List<Long> deletedDeviceIds) {
+    private List<IotSensor> convertToDomain(List<SensorDto> sensorDtos, String msgId) {
         List<IotSensor> domains = new ArrayList<>();
         List<NorthError> errors = new ArrayList<>();
         String path = "/api/data/sensorCommonList";
-        Map<String, List<IotSensor>> productDeviceMap = getAllDevice();
+
+        final SensorDto dto = sensorDtos.get(0);
+        final String orgCode = dto.getOrgCode();
+        SysOrg org = remoteOrgService.findByCode(orgCode, SecurityConstants.INNER);
+        if (org == null) {
+            errors.add(new NorthError(msgId, path, dto, "参数非法：机构orgCode不存在:" + orgCode));
+            return new ArrayList<>();
+        }
+        final long count = sensorDtos.stream().map(SensorDto::getOrgCode).distinct().count();
+        if (count > 1) {
+            errors.add(new NorthError(msgId, path, dto, "数据异常：完整数据内发现存在不同机构数据:" + orgCode));
+            return new ArrayList<>();
+        }
+        Map<String, List<IotSensor>> productDeviceMap = getAllDevice(org.getId());
         Set<String> uniqueSet = new HashSet<>();
         for (SensorDto sensorDto : sensorDtos) {
             NorthError error = null;
@@ -214,14 +242,7 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
                 continue;
             }
 
-            SysOrg org = remoteOrgService.findByCode(sensorDto.getOrgCode(), SecurityConstants.INNER);
-            if (org == null) {
-                error = new NorthError(msgId, path
-                        , sensorDto, "参数非法：机构orgCode不存在:" + sensorDto.getOrgCode());
-                errors.add(error);
-                continue;
-            }
-            String key = org.getId() + sensorDto.getDeviceCode();
+            String key = sensorDto.getDeviceCode();
             if (uniqueSet.contains(key)) {
                 errors.add(new NorthError(msgId, path
                         , sensorDto, "数据来源中，同机构中设备编号重复。"));
@@ -239,7 +260,6 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
             domain.setDeviceType(sensorDto.getDeviceType());
             domain.setDeleted(0);
             domain.setCreateTime(LocalDateTime.now());
-
             if (ObjectUtil.isNotEmpty(exists)) {
                 if (exists.size() > 1) {
                     errors.add(new NorthError(msgId, path, sensorDto, "同机构下设备编号重复，将取第一条"));
@@ -256,24 +276,21 @@ public class IotSensorServiceImpl extends ServiceImpl<IotSensorMapper, IotSensor
             domain.setSource(1);
             domains.add(domain);
         }
-
         northErrorService.saveErrorData(errors);
-
-
-        productDeviceMap.keySet().stream().filter(k -> !uniqueSet.contains(k)).forEach(k -> productDeviceMap.get(k).forEach(d -> deletedDeviceIds.add(d.getId())));
-
         return domains;
     }
 
-    private Map<String, List<IotSensor>> getAllDevice() {
+    private Map<String, List<IotSensor>> getAllDevice(Long orgId) {
         return baseMapper.selectList(new LambdaQueryWrapper<IotSensor>()
+                        .eq(ObjectUtil.isNotNull(orgId), IotSensor::getOrgId, orgId)
+                        .eq(IotSensor::getDeleted, 0)
                         .eq(IotSensor::getSource, 1))
                 .stream()
-                .collect(Collectors.groupingBy(c -> String.valueOf(c.getOrgId())));
+                .collect(Collectors.groupingBy(IotSensor::getDeviceCode));
     }
 
     private String convertDeviceType(String deviceType) {
-        if (ObjectUtil.isEmpty(deviceType)){
+        if (ObjectUtil.isEmpty(deviceType)) {
             return null;
         }
         return DictUtils.getDictLabel(DictConstants.SENSOR_DEVICE_TYPE, Integer.parseInt(deviceType));

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/util/RedisCheckRepeatDataUtil.java

@@ -106,7 +106,7 @@ public class RedisCheckRepeatDataUtil {
             allPageKeyList.add(everyPageKey);
         }
         //判断是否拿到所有的数据
-        final List<Boolean> list = allPageKeyList.stream().map(key -> redisTemplate.hasKey(key)).distinct().collect(Collectors.toList());
+        final List<Boolean> list = allPageKeyList.stream().map(redisTemplate::hasKey).distinct().collect(Collectors.toList());
         if (list.contains(Boolean.FALSE)) {
             // return ListUtil.empty();
         }

soc-modules/soc-modules-mediator/src/main/resources/mapper/IotSensorMapper.xml

@@ -1,5 +1,9 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.xunmei.mediator.api.mapper.IotSensorMapper">
-
+    <update id="updateDelByOrgId">
+        update iot_sensor
+        set deleted=1
+        where org_id = #{orgId}
+    </update>
 </mapper>