Merge branch 'data-zyj' into server-165

data-gateway/data-gateway-eg-kat/src/main/java/com/usky/gateway/domain/EgDeviceConfig.java

@@ -0,0 +1,19 @@
+package com.usky.gateway.domain;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class EgDeviceConfig implements Serializable {
+    private String deviceUuid;
+
+    private String deviceId;
+
+    private String deviceName;
+
+    private String ip;
+    private String port;
+
+    private String egNumber;
+}

data-gateway/data-gateway-eg-kat/src/main/java/com/usky/gateway/mapper/EgDeviceConfigMapper.java

@@ -0,0 +1,19 @@
+package com.usky.gateway.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.usky.gateway.domain.EgDeviceConfig;
+import com.usky.gateway.domain.GatewayDeviceConfig;
+import org.apache.ibatis.annotations.Mapper;
+import org.springframework.stereotype.Repository;
+
+/**
+ * <p>
+ * 用户信息表 Mapper 接口
+ * </p>
+ *
+ * @since 2022-04-21
+ */
+@Repository
+@Mapper
+public interface EgDeviceConfigMapper extends BaseMapper<EgDeviceConfig> {}  // 结合mybatis plus
+

data-gateway/data-gateway-eg-kat/src/main/java/com/usky/gateway/service/config/mqtt/MqttBaseConfig.java

@@ -25,8 +25,8 @@ public class MqttBaseConfig {
 	@Value("${mqtt.url}")
 	private String hostUrl;
 
-	@Value("${mqtt.sub-topics}")
-	private String msgTopic;
+//	@Value("${mqtt.sub-topics}")
+//	private String msgTopic;
 
 	@Value("${mqtt.keep-alive-interval}")
 	//心跳间隔

data-gateway/data-gateway-eg-kat/src/main/java/com/usky/gateway/service/config/mqtt/MqttInConfig.java

@@ -1,5 +1,10 @@
 package com.usky.gateway.service.config.mqtt;
 
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.usky.gateway.domain.EgDeviceConfig;
+import com.usky.gateway.domain.GatewayDeviceConfig;
+import com.usky.gateway.mapper.EgDeviceConfigMapper;
 import lombok.Data;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -24,6 +29,8 @@ public class MqttInConfig {
 
     @Autowired
     private MqttBaseConfig mqttBaseConfig;
+    @Autowired
+    private EgDeviceConfigMapper egDeviceConfigMapper;
 
     public static final String CHANNEL_NAME_INPUT = "mqttInputChannel";
 
@@ -40,7 +47,15 @@ public class MqttInConfig {
      */
     @Bean
     public MessageProducer inbound() {
-        String[] tops = mqttBaseConfig.getMsgTopic().split(",");
+//        String[] tops = mqttBaseConfig.getMsgTopic().split(",");
+        LambdaQueryWrapper<EgDeviceConfig> queryWrapper = Wrappers.lambdaQuery();
+        queryWrapper.select(EgDeviceConfig::getDeviceUuid);
+        List<EgDeviceConfig> list = egDeviceConfigMapper.selectList(queryWrapper);
+        String[] tops = new String[list.size()];
+        for (int i = 0; i < list.size(); i++) {
+            tops[i] = "/"+list.get(i).getDeviceUuid()+"/control";
+        }
+
         String clientId = "gateway-eg-kat-mqtt-in-" + System.currentTimeMillis();
         MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,
                 mqttBaseConfig.mqttClientFactory(), tops);

data-gateway/data-gateway-eg-kat/src/main/java/com/usky/gateway/service/mqtt/control/control.java

@@ -4,7 +4,9 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.usky.gateway.domain.EgDeviceConfig;
 import com.usky.gateway.domain.GatewayDeviceConfig;
+import com.usky.gateway.mapper.EgDeviceConfigMapper;
 import com.usky.gateway.mapper.GatewayDeviceConfigMapper;
 import com.usky.gateway.service.config.mqtt.MqttOutConfig;
 import com.usky.gateway.service.config.udp.UdpBaseConfig;
@@ -29,11 +31,6 @@ import java.util.Map;
 @Configuration
 @Data
 public class control implements MqttStrategy {
-//    @Value("${udp.deviceIp}")
-//    private String deviceIp;
-//
-//    @Value("${udp.sendingPort}")
-//    private Integer sendingPort;
 
     @Autowired
     private UdpUtil udpUtil;
@@ -43,32 +40,30 @@ public class control implements MqttStrategy {
     private MqttOutConfig.MqttGateway mqttGateway;
 
     @Autowired
-    private GatewayDeviceConfigMapper gatewayDeviceConfigMapper;
+    private EgDeviceConfigMapper egDeviceConfigMapper;
 
     //处理下发命令消息，下发命令控制设备
     public String disposeMessage(MqttBaseVO mqttBaseVO) {
         String[] topics = mqttBaseVO.getTopic().split("/");
-        String productCode = topics[1];
+        String deviceUuid = topics[1];
         JSONObject obj_data = JSONObject.parseObject(mqttBaseVO.getData().toString());
         Integer commandId = Integer.parseInt(obj_data.get("id").toString());
         Object params = JSONObject.toJSONString(obj_data.get("params"));
         JSONObject params_data = JSON.parseObject(params.toString());
-        String deviceId = params_data.getString("deviceId");
         String commandCode = params_data.getString("commandCode");
         String commandValue = params_data.getString("commandValue");
 
         //开门   17 40 00 00 E0 F4 4D 0D 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00
         if(commandCode.equals("door_onoff")){
             //通过查询db文件获取对应udpIp udpPort 目标设备id 门号
-            LambdaQueryWrapper<GatewayDeviceConfig> queryWrapper = Wrappers.lambdaQuery();
-            queryWrapper.select(GatewayDeviceConfig::getIp,GatewayDeviceConfig::getPort,GatewayDeviceConfig::getParam1,GatewayDeviceConfig::getParam2)
-                    .eq(GatewayDeviceConfig::getProductCode,productCode)
-                    .eq(GatewayDeviceConfig::getDeviceId,deviceId);
-            GatewayDeviceConfig one = gatewayDeviceConfigMapper.selectOne(queryWrapper);
+            LambdaQueryWrapper<EgDeviceConfig> queryWrapper = Wrappers.lambdaQuery();
+            queryWrapper.select(EgDeviceConfig::getDeviceId,EgDeviceConfig::getIp,EgDeviceConfig::getPort,EgDeviceConfig::getEgNumber)
+                    .eq(EgDeviceConfig::getDeviceUuid,deviceUuid);
+            EgDeviceConfig one = egDeviceConfigMapper.selectOne(queryWrapper);
             if(one != null){
-                String devId = one.getParam1();
+                String devId = one.getDeviceId();
                 byte[] deviceByte = byteUtil.toLH(Integer.parseInt(devId));
-                int doorNum = Integer.parseInt(one.getParam2());
+                int doorNum = Integer.parseInt(one.getEgNumber());
                 byte bDoorNum = (byte)doorNum;
                 byte[] requestBytes = new byte[] { 0x17, 0x40, 0x00, 0x00, deviceByte[0], deviceByte[1], deviceByte[2], deviceByte[3], bDoorNum, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
                 byteUtil.printBytes(requestBytes);

data-gateway/data-gateway-eg-kat/src/main/resources/application.yml

@@ -33,7 +33,8 @@ spring:
       max-request-size: 15MB
   datasource:
     name: sqlite3
-    url: jdbc:sqlite::resource:db/yt_gateway.db
+    #url: jdbc:sqlite::resource:db/yt_gateway.db
+    url: jdbc:sqlite:db/yt_gateway.db
     driver-class-name: org.sqlite.JDBC
     username:
     password:
@@ -49,7 +50,7 @@ mqtt:
 #  url: tcp://192.168.123.165:1883
 #  username: admin
   password: usky
-  sub-topics: /502_KAT/+/control
+#  sub-topics: /502_KAT/+/control
   url: tcp://47.98.201.73:1883
   username: usky
 # 和嵌入式udp通信的发送端口和监听端口地址

data-transfer/data-transfer-api/src/main/java/com/usky/transfer/RemoteTransferService.java

@@ -24,5 +24,5 @@ public interface RemoteTransferService {
      * 下发单个设备控制命令
      */
     @PostMapping("/deviceControl")
-    Map<String,Object> deviceControl(@RequestParam("productCode") String productCode, @RequestParam(value = "deviceId",required = false) String deviceId, @RequestParam("commandStr") String commandStr,@RequestParam(value = "tenantId") Integer tenantId, @RequestParam(value = "userId") Long userId, @RequestParam(value = "userName") String userName);
+    Map<String,Object> deviceControl(@RequestParam("productCode") String productCode, @RequestParam(value = "deviceUuid",required = false) String deviceUuid, @RequestParam("commandStr") String commandStr,@RequestParam(value = "tenantId") Integer tenantId, @RequestParam(value = "userId") Long userId, @RequestParam(value = "userName") String userName);
 }

data-transfer/data-transfer-api/src/main/java/com/usky/transfer/domain/DeviceDataWriteVO.java

@@ -27,7 +27,7 @@ public class DeviceDataWriteVO implements Serializable {
     /**
      * 数据上报时间
      */
-    private long timestamp;
+    private Long timestamp;
 
     /**
      * 标签集合

data-transfer/data-transfer-api/src/main/java/com/usky/transfer/domain/MqttDeviceDataVO.java

@@ -0,0 +1,18 @@
+package com.usky.transfer.domain;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class MqttDeviceDataVO implements Serializable {
+    /**
+     * topic
+     */
+    private String topic;
+
+    /**
+     * 设备数据体
+     */
+    private String deviceData;
+}

data-transfer/data-transfer-api/src/main/java/com/usky/transfer/factory/RemoteTransferFallbackFactory.java

@@ -36,7 +36,7 @@ public class RemoteTransferFallbackFactory implements FallbackFactory<RemoteTran
                 throw new BusinessException(throwable.getMessage());
             }
             @Override
-            public Map<String,Object> deviceControl(String productCode, String deviceId, String commandStr,Integer tenantId, Long userId, String userName) {
+            public Map<String,Object> deviceControl(String productCode, String deviceUuid, String commandStr,Integer tenantId, Long userId, String userName) {
                 throw new BusinessException(throwable.getMessage());
             }
         };

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/controller/api/DataTransferControllerApi.java

@@ -29,7 +29,7 @@ public class DataTransferControllerApi implements RemoteTransferService {
      * 下发单个设备控制命令
      */
     @Override
-    public Map<String,Object> deviceControl(String productCode, String deviceId, String commandStr,Integer tenantId, Long userId, String userName){
-        return queryInfluxdbDataService.deviceControl(productCode, deviceId, commandStr, tenantId, userId, userName);
+    public Map<String,Object> deviceControl(String productCode, String deviceUuid, String commandStr,Integer tenantId, Long userId, String userName){
+        return queryInfluxdbDataService.deviceControl(productCode, deviceUuid, commandStr, tenantId, userId, userName);
     }
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/controller/web/QueryDeviceDataController.java

@@ -201,12 +201,12 @@ public class QueryDeviceDataController {
     @GetMapping("deviceCommandRecord")
     public ApiResult<CommonPage<DmpDeviceCommand>> deviceCommandRecord(@RequestParam(value = "commandStatus",required = false) Integer commandStatus,
                                                                        @RequestParam(value = "productCode",required = false) String productCode,
-                                                                       @RequestParam(value = "deviceId",required = false) String deviceId,
+                                                                       @RequestParam(value = "deviceUuid",required = false) String deviceUuid,
                                                                        @RequestParam(value = "userName",required = false) String userName,
                                                                        @RequestParam(value = "startTime",required = false) String startTime,
                                                                        @RequestParam(value = "endTime",required = false) String endTime,
                                                                        @RequestParam(value = "pageNum", required = false, defaultValue = "1") Integer pageNum,
                                                                        @RequestParam(value = "pageSize", required = false, defaultValue = "10") Integer pageSize){
-        return ApiResult.success(dmpDeviceCommandService.deviceCommandRecord(commandStatus,productCode,deviceId,userName,startTime,endTime,pageNum,pageSize));
+        return ApiResult.success(dmpDeviceCommandService.deviceCommandRecord(commandStatus,productCode,deviceUuid,userName,startTime,endTime,pageNum,pageSize));
     }
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/controller/web/SendMessageController.java

@@ -2,6 +2,7 @@ package com.usky.transfer.controller.web;
 
 import com.usky.common.core.bean.ApiResult;
 import com.usky.transfer.domain.DeviceDataWriteVO;
+import com.usky.transfer.domain.MqttDeviceDataVO;
 import com.usky.transfer.service.QueryInfluxdbDataService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
@@ -28,4 +29,15 @@ public class SendMessageController {
         return ApiResult.success(queryInfluxdbDataService.sendDeviceDataToMQ(writeVO));
     }
 
+    /**
+     * MQTT设备数据上报调试
+     * @param mqttDeviceDataVO
+     * @return
+     */
+    @PostMapping("/mqttDeviceData")
+    public ApiResult<Map<String,Object>> mqttDeviceData(@RequestBody MqttDeviceDataVO mqttDeviceDataVO){
+        return ApiResult.success(queryInfluxdbDataService.mqttDeviceData(mqttDeviceDataVO));
+    }
+
+
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/domain/DmpDeviceCommand.java

@@ -33,9 +33,9 @@ public class DmpDeviceCommand implements Serializable {
     private String productCode;
 
     /**
-     * 设备编码
+     * 设备Uuid
      */
-    private String deviceId;
+    private String deviceUuid;
 
     /**
      * 下发命令内容

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/DmpDeviceCommandService.java

@@ -15,5 +15,5 @@ import org.springframework.web.bind.annotation.RequestParam;
  */
 public interface DmpDeviceCommandService extends CrudService<DmpDeviceCommand> {
 
-    CommonPage<DmpDeviceCommand> deviceCommandRecord(Integer commandStatus,String productCode,String deviceId,String userName,String startTime,String endTime,Integer pageNum,Integer pageSize);
+    CommonPage<DmpDeviceCommand> deviceCommandRecord(Integer commandStatus,String productCode,String deviceUuid,String userName,String startTime,String endTime,Integer pageNum,Integer pageSize);
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/DmpProductService.java

@@ -17,8 +17,8 @@ import java.util.Map;
  */
 public interface DmpProductService extends CrudService<DmpProduct> {
     void deleteProductCache();
-    void deleteDeviceCache();
+    void deleteDeviceCache(String productCode);
 
     Map<String, ProductMapVO> getProductMap();
-    Map<String, DeviceMapVO> getDeviceMap();
+    Map<String, DeviceMapVO> getDeviceMap(String productCode);
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/QueryInfluxdbDataService.java

@@ -2,6 +2,7 @@ package com.usky.transfer.service;
 
 import com.usky.transfer.domain.*;
 import com.usky.common.mybatis.core.CrudService;
+import org.springframework.web.bind.annotation.RequestBody;
 
 import java.util.List;
 import java.util.Map;
@@ -18,6 +19,8 @@ public interface QueryInfluxdbDataService extends CrudService<QueryInfluxdbData>
 
     Map<String,Object> sendDeviceDataToMQ(DeviceDataWriteVO writeVO);
 
-    Map<String,Object> deviceControl(String productCode, String deviceId, String commandStr,Integer tenantId, Long userId, String userName);
+    Map<String,Object> mqttDeviceData(MqttDeviceDataVO mqttDeviceDataVO);
+
+    Map<String,Object> deviceControl(String productCode, String deviceUuid, String commandStr,Integer tenantId, Long userId, String userName);
 
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/impl/DmpDeviceCommandServiceImpl.java

@@ -25,12 +25,12 @@ import org.springframework.stereotype.Service;
 public class DmpDeviceCommandServiceImpl extends AbstractCrudService<DmpDeviceCommandMapper, DmpDeviceCommand> implements DmpDeviceCommandService {
 
     @Override
-    public CommonPage<DmpDeviceCommand> deviceCommandRecord(Integer commandStatus,String productCode,String deviceId, String userName, String startTime, String endTime, Integer pageNum, Integer pageSize){
+    public CommonPage<DmpDeviceCommand> deviceCommandRecord(Integer commandStatus,String productCode,String deviceUuid, String userName, String startTime, String endTime, Integer pageNum, Integer pageSize){
         IPage<DmpDeviceCommand> page = new Page<>(pageNum,pageSize);
         LambdaQueryWrapper<DmpDeviceCommand> queryWrapper = Wrappers.lambdaQuery();
         queryWrapper.eq(commandStatus != null,DmpDeviceCommand::getCommandStatus,commandStatus)
                 .like(StringUtils.isNotBlank(productCode),DmpDeviceCommand::getProductCode,productCode)
-                .like(StringUtils.isNotBlank(deviceId),DmpDeviceCommand::getDeviceId,deviceId)
+                .like(StringUtils.isNotBlank(deviceUuid),DmpDeviceCommand::getDeviceUuid,deviceUuid)
                 .like(StringUtils.isNotBlank(userName),DmpDeviceCommand::getUserName,userName)
                 .between(StringUtils.isNotBlank(startTime)&&StringUtils.isNotBlank(endTime),DmpDeviceCommand::getCreatedTime,startTime,endTime)
                 .eq(DmpDeviceCommand::getTenantId, SecurityUtils.getTenantId())

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/impl/DmpProductServiceImpl.java

@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.usky.common.mybatis.core.AbstractCrudService;
+import com.usky.common.security.utils.SecurityUtils;
 import com.usky.transfer.domain.DmpDevice;
 import com.usky.transfer.domain.DmpProduct;
 import com.usky.transfer.mapper.DmpProductMapper;
@@ -42,8 +43,8 @@ public class DmpProductServiceImpl extends AbstractCrudService<DmpProductMapper,
     }
 
     //清除设备缓存
-    @CacheEvict(cacheNames = "deviceList")
-    public void deleteDeviceCache(){
+    @CacheEvict(cacheNames = "deviceList",key = "#productCode")
+    public void deleteDeviceCache(String productCode){
         System.out.println("deleteDeviceCache");
     }
 
@@ -72,11 +73,12 @@ public class DmpProductServiceImpl extends AbstractCrudService<DmpProductMapper,
         return productMap;
     }
 
-    @Cacheable(cacheNames = "deviceList",sync = true)
-    public Map<String, DeviceMapVO> getDeviceMap(){
+    @Cacheable(cacheNames = "deviceList",key = "#productCode",sync = true)
+    public Map<String, DeviceMapVO> getDeviceMap(String productCode){
         Map<String,DeviceMapVO> deviceMap = new HashMap<>();
         LambdaQueryWrapper<DmpDevice> queryWrapper = Wrappers.lambdaQuery();
         queryWrapper.eq(DmpDevice::getDeleteFlag,0)
+                .eq(DmpDevice::getProductCode,productCode)
                 .orderByDesc(DmpDevice::getId);
         List<DmpDevice> list = dmpDeviceService.list(queryWrapper);
         if(CollectionUtils.isNotEmpty(list)){

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/impl/QueryInfluxdbDataServiceImpl.java

@@ -56,15 +56,15 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
     private MqttOutConfig.MqttGateway mqttGateway;
 
     @Override
-    public Map<String,Object> deviceControl(String productCode, String deviceId, String commandStr,Integer tenantId, Long userId, String userName){
+    public Map<String,Object> deviceControl(String productCode, String deviceUuid, String commandStr,Integer tenantId, Long userId, String userName){
         Map<String,Object> rec_map = new HashMap<>();
 
 
         //存储下发设备控制命令到数据库表中
         DmpDeviceCommand command = new DmpDeviceCommand();
         command.setProductCode(productCode);
-        if(StringUtils.isNotBlank(deviceId)){
-            command.setDeviceId(deviceId);
+        if(StringUtils.isNotBlank(deviceUuid)){
+            command.setDeviceUuid(deviceUuid);
         }
 
         command.setCommandContent(commandStr);
@@ -86,7 +86,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
         dmpDeviceCommandService.updateById(command);
         //推送下发设备控制mqtt
         if(StringUtils.isNotBlank(commandStr)){
-            String topic = "/"+productCode+"/"+deviceId+"/control";
+            String topic = "/"+deviceUuid+"/control";
             mqttGateway.sendToMqtt(topic,dataJson.toJSONString());
         }
 
@@ -106,6 +106,8 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
             if((endTimeStamp - startTimeStamp) >= 3){ //请求超时3秒，返回失败
                 rec_map.put("code",-1);
                 rec_map.put("message","下发命令失败");
+
+                dmpDeviceCommandService.lambdaUpdate().set(DmpDeviceCommand::getCommandStatus,2).eq(DmpDeviceCommand::getId,commandId).update();
                 break;
             }
         }
@@ -137,7 +139,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
 
             String productCode = writeVO.getProductCode();
             String deviceUUId = writeVO.getDeviceUUId();
-            String deviceId = tags.get("deviceId");
+            String deviceId = tags.get("device_id");
 
             //判断上报数据对应产品是否注册，如未注册则为非法
             Map<String,ProductMapVO> productMapList = dmpProductService.getProductMap();
@@ -158,7 +160,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
             }
 
             //判断上报数据设备是否已注册(要判断注册过的设备是不是属于本产品的)，未注册自动注册
-            Map<String,DeviceMapVO> deviceMapList = dmpProductService.getDeviceMap();
+            Map<String,DeviceMapVO> deviceMapList = dmpProductService.getDeviceMap(productCode);
             if(!deviceMapList.containsKey(deviceId)){
                 DmpDevice dmpDeviceInfo = new DmpDevice();
                 ProductMapVO productMapVO = productMapList.get(productCode);
@@ -173,6 +175,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
                 dmpDeviceInfo.setServiceStatus(1);
                 dmpDeviceInfo.setDeviceUuid(UUIDUtils.uuid());
                 dmpDeviceService.save(dmpDeviceInfo);
+                dmpProductService.deleteDeviceCache(productCode);
 
                 deviceUUId = dmpDeviceInfo.getDeviceUuid();
 
@@ -184,7 +187,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
                 dmpDeviceStatus.setProductCode(dmpDeviceInfo.getProductCode());
                 dmpDeviceStatusService.save(dmpDeviceStatus);
 
-                dmpProductService.deleteDeviceCache();
+                deviceMapList = dmpProductService.getDeviceMap(productCode);
             }else if(deviceMapList.containsKey(deviceId)){
                 LambdaQueryWrapper<DmpDevice> queryWrapper = Wrappers.lambdaQuery();
                 queryWrapper.eq(DmpDevice::getDeleteFlag,0)
@@ -205,6 +208,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
                     dmpDeviceInfo.setServiceStatus(1);
                     dmpDeviceInfo.setDeviceUuid(UUIDUtils.uuid());
                     dmpDeviceService.save(dmpDeviceInfo);
+                    dmpProductService.deleteDeviceCache(productCode);
 
                     deviceUUId = dmpDeviceInfo.getDeviceUuid();
 
@@ -216,7 +220,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
                     dmpDeviceStatus.setProductCode(dmpDeviceInfo.getProductCode());
                     dmpDeviceStatusService.save(dmpDeviceStatus);
 
-                    dmpProductService.deleteDeviceCache();
+                    deviceMapList = dmpProductService.getDeviceMap(productCode);
                 }
             }
 
@@ -226,6 +230,7 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
                     String deviceId1 = map.getKey();
                     if((productCode.equals(productCode1)) && (deviceId.equals(deviceId1))){
                         deviceUUId = map.getValue().getDeviceUuid();
+                        break;
                     }
                 }
             }
@@ -242,4 +247,20 @@ public class QueryInfluxdbDataServiceImpl extends AbstractCrudService<QueryInflu
         return rec_map;
     }
 
+    @Override
+    public Map<String,Object> mqttDeviceData(MqttDeviceDataVO mqttDeviceDataVO){
+        Map<String,Object> rec_map = new HashMap<>();
+
+        String topic = mqttDeviceDataVO.getTopic();
+        String payload = mqttDeviceDataVO.getDeviceData();
+
+        JSONObject deviceDataJson = JSONObject.parseObject(payload);
+
+        mqttGateway.sendToMqtt(topic, deviceDataJson.toJSONString());
+
+        rec_map.put("code",200);
+        rec_map.put("message","发送成功");
+
+        return rec_map;
+    }
 }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/listener/MqttListener.java

@@ -52,6 +52,18 @@ public class MqttListener {
                     mqttBaseVO.setDescribe("controlResponse");
                     mqttBaseVO.setData(payload);
                 }
+                else if(topic.indexOf("add") != -1){
+                    mqttBaseVO.setDescribe("add");
+                    mqttBaseVO.setData(payload);
+                }
+                else if(topic.indexOf("data-collector") != -1){
+                    mqttBaseVO.setDescribe("dataCollector");
+                    mqttBaseVO.setData(payload);
+                }
+                else if(topic.indexOf("alarm") != -1){
+                    mqttBaseVO.setDescribe("alarm");
+                    mqttBaseVO.setData(payload);
+                }
                 //统一处理数据
                 simpleContext.getResource(mqttBaseVO);
             }

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/mqtt/add/Add.java

@@ -0,0 +1,122 @@
+package com.usky.transfer.service.mqtt.add;
+
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.nacos.shaded.com.google.gson.JsonObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.usky.common.core.util.UUIDUtils;
+import com.usky.transfer.domain.DmpDevice;
+import com.usky.transfer.domain.DmpDeviceStatus;
+import com.usky.transfer.domain.DmpProduct;
+import com.usky.transfer.service.DmpDeviceService;
+import com.usky.transfer.service.DmpDeviceStatusService;
+import com.usky.transfer.service.DmpProductService;
+import com.usky.transfer.service.mqtt.MqttStrategy;
+import com.usky.transfer.service.vo.DeviceMapVO;
+import com.usky.transfer.service.vo.MqttBaseVO;
+import com.usky.transfer.service.vo.ProductMapVO;
+import lombok.extern.slf4j.Slf4j;
+import org.checkerframework.checker.units.qual.A;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+@Service("add")
+public class Add implements MqttStrategy {
+    @Autowired
+    private DmpProductService dmpProductService;
+
+    @Autowired
+    private DmpDeviceService dmpDeviceService;
+
+    @Autowired
+    private DmpDeviceStatusService dmpDeviceStatusService;
+
+    public String disposeMessage(MqttBaseVO mqttBaseVO){
+        String deviceInfoStr = mqttBaseVO.getData().toString();
+        JSONObject deviceInfoJson = JSONObject.parseObject(deviceInfoStr);
+        String productCode = deviceInfoJson.get("productCode").toString();
+        String deviceId = deviceInfoJson.get("deviceId").toString();
+
+        //判断上报数据对应产品是否注册，如未注册则为非法
+        Map<String, ProductMapVO> productMapList = dmpProductService.getProductMap();
+        if(!productMapList.containsKey(productCode)){
+            //通过查询数据库再确认下产品是否注册
+            LambdaQueryWrapper<DmpProduct> queryWrapper = Wrappers.lambdaQuery();
+            queryWrapper.eq(DmpProduct::getDeleteFlag,0)
+                    .eq(DmpProduct::getProductCode,productCode);
+            DmpProduct one = dmpProductService.getOne(queryWrapper);
+            if(one == null){
+                log.error(productCode+" 产品未注册");
+                return null;
+            }else{
+                dmpProductService.deleteProductCache();
+            }
+        }
+
+        //判断上报数据设备是否已注册(要判断注册过的设备是不是属于本产品的)，未注册自动注册
+        Map<String, DeviceMapVO> deviceMapList = dmpProductService.getDeviceMap(productCode);
+        if(!deviceMapList.containsKey(deviceId)){
+            DmpDevice dmpDeviceInfo = new DmpDevice();
+            ProductMapVO productMapVO = productMapList.get(productCode);
+            dmpDeviceInfo.setDeviceId(deviceId);
+            dmpDeviceInfo.setDeviceName("");
+            dmpDeviceInfo.setDeviceType(productMapVO.getDeviceType());
+            dmpDeviceInfo.setProductId(productMapVO.getProductId());
+            dmpDeviceInfo.setProductCode(productCode);
+            dmpDeviceInfo.setCreatedBy(productMapVO.getCreatedBy());
+            dmpDeviceInfo.setCreatedTime(LocalDateTime.now());
+            dmpDeviceInfo.setTenantId(productMapVO.getTenantId());
+            dmpDeviceInfo.setServiceStatus(1);
+            dmpDeviceInfo.setDeviceUuid(UUIDUtils.uuid());
+            dmpDeviceService.save(dmpDeviceInfo);
+
+            DmpDeviceStatus dmpDeviceStatus = new DmpDeviceStatus();
+            dmpDeviceStatus.setDeviceId(dmpDeviceInfo.getDeviceId());
+            dmpDeviceStatus.setProductId(dmpDeviceInfo.getProductId());
+            dmpDeviceStatus.setDeviceStatus(2);
+            dmpDeviceStatus.setLastOfflineTime(LocalDateTime.now());
+            dmpDeviceStatus.setProductCode(dmpDeviceInfo.getProductCode());
+            dmpDeviceStatusService.save(dmpDeviceStatus);
+
+            dmpProductService.deleteDeviceCache(productCode);
+        }else if(deviceMapList.containsKey(deviceId)){
+            LambdaQueryWrapper<DmpDevice> queryWrapper = Wrappers.lambdaQuery();
+            queryWrapper.eq(DmpDevice::getDeleteFlag,0)
+                    .eq(DmpDevice::getProductCode,productCode)
+                    .eq(DmpDevice::getDeviceId,deviceId);
+            DmpDevice one = dmpDeviceService.getOne(queryWrapper);
+            if(one == null){
+                DmpDevice dmpDeviceInfo = new DmpDevice();
+                ProductMapVO productMapVO = productMapList.get(productCode);
+                dmpDeviceInfo.setDeviceId(deviceId);
+                dmpDeviceInfo.setDeviceName("");
+                dmpDeviceInfo.setDeviceType(productMapVO.getDeviceType());
+                dmpDeviceInfo.setProductId(productMapVO.getProductId());
+                dmpDeviceInfo.setProductCode(productCode);
+                dmpDeviceInfo.setCreatedBy(productMapVO.getCreatedBy());
+                dmpDeviceInfo.setCreatedTime(LocalDateTime.now());
+                dmpDeviceInfo.setTenantId(productMapVO.getTenantId());
+                dmpDeviceInfo.setServiceStatus(1);
+                dmpDeviceInfo.setDeviceUuid(UUIDUtils.uuid());
+                dmpDeviceService.save(dmpDeviceInfo);
+
+                DmpDeviceStatus dmpDeviceStatus = new DmpDeviceStatus();
+                dmpDeviceStatus.setDeviceId(dmpDeviceInfo.getDeviceId());
+                dmpDeviceStatus.setProductId(dmpDeviceInfo.getProductId());
+                dmpDeviceStatus.setDeviceStatus(2);
+                dmpDeviceStatus.setLastOfflineTime(LocalDateTime.now());
+                dmpDeviceStatus.setProductCode(dmpDeviceInfo.getProductCode());
+                dmpDeviceStatusService.save(dmpDeviceStatus);
+
+                dmpProductService.deleteDeviceCache(productCode);
+            }
+        }
+
+        return null;
+    }
+}

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/mqtt/alarm/alarm.java

@@ -0,0 +1,40 @@
+package com.usky.transfer.service.mqtt.alarm;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.usky.common.core.util.JsonUtils;
+import com.usky.transfer.domain.DeviceDataWriteVO;
+import com.usky.transfer.service.QueryInfluxdbDataService;
+import com.usky.transfer.service.mqtt.MqttStrategy;
+import com.usky.transfer.service.rocketmq.MyProducer;
+import com.usky.transfer.service.utils.HttpClientUtils;
+import com.usky.transfer.service.vo.MqttBaseVO;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author zyj
+ * @date 2025/1/13 15:07
+ */
+@Service("alarm")
+public class alarm implements MqttStrategy {
+    @Value("${alarm.url}")
+    private String alarmUrl;
+
+    public String disposeMessage(MqttBaseVO mqttBaseVO) {
+        String data = mqttBaseVO.getData().toString();
+        JSONObject jsonObject = JSONObject.parseObject(data);
+        String resultString = HttpClientUtils.doPostJson(alarmUrl,jsonObject.toJSONString());
+
+        System.out.println("resultString: "+resultString);
+
+        return null;
+    }
+
+
+}

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/mqtt/datacollector/DataCollector.java

@@ -0,0 +1,58 @@
+package com.usky.transfer.service.mqtt.datacollector;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.usky.common.core.util.JsonUtils;
+import com.usky.transfer.domain.DeviceDataWriteVO;
+import com.usky.transfer.service.QueryInfluxdbDataService;
+import com.usky.transfer.service.mqtt.MqttStrategy;
+import com.usky.transfer.service.rocketmq.MyProducer;
+import com.usky.transfer.service.vo.MqttBaseVO;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author zyj
+ * @date 2022/12/6 15:07
+ */
+@Service("dataCollector")
+public class DataCollector implements MqttStrategy {
+    @Resource
+    private MyProducer myProducer;
+    @Autowired
+    private QueryInfluxdbDataService queryInfluxdbDataService;
+
+    public String disposeMessage(MqttBaseVO mqttBaseVO) {
+        Map<String, String> tags = new HashMap<>();
+        Map<String, Object> fields = new HashMap<>();
+        Map map_data = JsonUtils.fromJson(mqttBaseVO.getData().toString(), Map.class);
+        String productCode = map_data.get("product_id").toString().toLowerCase();
+        long timestamp = Long.valueOf(map_data.get("timestamp").toString())*1000L+1L;
+        String deviceId = map_data.get("device_id").toString();
+
+        tags.put("device_id",deviceId);
+
+        Object met = JSONObject.toJSONString(map_data.get("metrics"));
+        JSONObject metrics = JSON.parseObject(met.toString());
+        for(String entry : metrics.keySet()){
+            fields.put(entry.toLowerCase(),metrics.get(entry));
+        }
+
+        DeviceDataWriteVO deviceDataWriteVO = new DeviceDataWriteVO();
+        deviceDataWriteVO.setDeviceUUId("");
+        deviceDataWriteVO.setProductCode(productCode);
+        deviceDataWriteVO.setTimestamp(timestamp);
+        deviceDataWriteVO.setTags(tags);
+        deviceDataWriteVO.setMetrics(metrics);
+
+        queryInfluxdbDataService.sendDeviceDataToMQ(deviceDataWriteVO);
+
+        return null;
+    }
+
+
+}

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/mqtt/info/Info.java

@@ -19,10 +19,7 @@ import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 /**
  * @author zyj
@@ -39,11 +36,9 @@ public class Info implements MqttStrategy {
         Map<String, String> tags = new HashMap<>();
         Map<String, Object> fields = new HashMap<>();
         Map map_data = JsonUtils.fromJson(mqttBaseVO.getData().toString(), Map.class);
-        String deviceId = map_data.get("device_id").toString();
-        String productCode = map_data.get("product_code").toString().toLowerCase();
-        Long timestamp = Long.valueOf(map_data.get("timestamp").toString());
+        String productCode = map_data.get("productCode").toString().toLowerCase();
 
-        String tableName = deviceId;
+        long timestamp = Long.valueOf(map_data.get("timestamp").toString());
 
         Object tg = JSONObject.toJSONString(map_data.get("tags"));
         JSONObject tag = JSON.parseObject(tg.toString());
@@ -58,9 +53,9 @@ public class Info implements MqttStrategy {
         }
 
         DeviceDataWriteVO deviceDataWriteVO = new DeviceDataWriteVO();
-        deviceDataWriteVO.setDeviceUUId(deviceId);
+        deviceDataWriteVO.setDeviceUUId("");
         deviceDataWriteVO.setProductCode(productCode);
-        deviceDataWriteVO.setTimestamp(LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli());
+        deviceDataWriteVO.setTimestamp(timestamp);
         deviceDataWriteVO.setTags(tags);
         deviceDataWriteVO.setMetrics(metrics);
 

data-transfer/data-transfer-biz/src/main/java/com/usky/transfer/service/utils/HttpClientUtils.java

@@ -0,0 +1,137 @@
+package com.usky.transfer.service.utils;
+
+import org.apache.http.NameValuePair;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.util.EntityUtils;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ 1. 跨域请求工具类
+ */
+public class HttpClientUtils {
+
+    public static String doGet(String url, Map<String, String> param) {
+
+        // 创建Httpclient对象
+        CloseableHttpClient httpclient = HttpClients.createDefault();
+
+        String resultString = "";
+        CloseableHttpResponse response = null;
+        try {
+            // 创建uri
+            URIBuilder builder = new URIBuilder(url);
+            if (param != null) {
+                for (String key : param.keySet()) {
+                    builder.addParameter(key, param.get(key));
+                }
+            }
+            URI uri = builder.build();
+
+            // 创建http GET请求
+            HttpGet httpGet = new HttpGet(uri);
+
+            // 执行请求
+            response = httpclient.execute(httpGet);
+            // 判断返回状态是否为200
+            if (response.getStatusLine().getStatusCode() == 200) {
+                resultString = EntityUtils.toString(response.getEntity(), "UTF-8");
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            try {
+                if (response != null) {
+                    response.close();
+                }
+                httpclient.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+        return resultString;
+    }
+
+    public static String doGet(String url) {
+        return doGet(url, null);
+    }
+
+    public static String doPost(String url, Map<String, Object> param) {
+        // 创建Httpclient对象
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+        CloseableHttpResponse response = null;
+        String resultString = "";
+        try {
+            // 创建Http Post请求
+            HttpPost httpPost = new HttpPost(url);
+            httpPost.setHeader("Content-Type", "charset=utf-8");
+            // 创建参数列表
+            if (param != null) {
+                List<NameValuePair> paramList = new ArrayList<>();
+                for (String key : param.keySet()) {
+                    paramList.add(new BasicNameValuePair(key, param.get(key).toString()));
+                }
+                // 模拟表单
+                UrlEncodedFormEntity entity = new UrlEncodedFormEntity(paramList);
+                httpPost.setEntity(entity);
+            }
+            // 执行http请求
+            response = httpClient.execute(httpPost);
+            resultString = EntityUtils.toString(response.getEntity(), "utf-8");
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            try {
+                response.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+
+        return resultString;
+    }
+
+    public static String doPost(String url) {
+        return doPost(url, null);
+    }
+
+    public static String doPostJson(String url, String json) {
+        // 创建Httpclient对象
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+        CloseableHttpResponse response = null;
+        String resultString = "";
+        try {
+            // 创建Http Post请求
+            HttpPost httpPost = new HttpPost(url);
+            // 创建请求内容
+            StringEntity entity = new StringEntity(json, ContentType.APPLICATION_JSON);
+            httpPost.setEntity(entity);
+            // 执行http请求
+            response = httpClient.execute(httpPost);
+            resultString = EntityUtils.toString(response.getEntity(), "utf-8");
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            try {
+                response.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+
+        return resultString;
+    }
+}

data-transfer/data-transfer-biz/src/main/resources/logback.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <configuration scan="true" scanPeriod="60 seconds" debug="false">
     <!-- 日志存放路径 -->
-	<property name="log.path" value="/var/log/uskycloud/usky-demo" />
+	<property name="log.path" value="/var/log/uskycloud/data-transfer" />
    <!-- 日志输出格式 -->
 	<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n" />