数据处理

iot-framework/iot-starter-redis/src/main/java/com/middle/platform/redis/constant/CacheConstant.java

@@ -8,4 +8,5 @@ public interface CacheConstant {
     String USER_CACHE = "user:%s";
     String DICT_CACHE = "dict:%s";
     String DICT_ITEM_CACHE = "dictItem:%s:%s";
+    String TOPIC_CACHE = "topic:%s";
 }

iot-module/iot-module-auth/iot-module-auth-biz/src/main/java/com/middle/platform/auth/biz/AuthApplication.java

@@ -2,8 +2,10 @@ package com.middle.platform.auth.biz;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 @SpringBootApplication
+@EnableTransactionManagement
 public class AuthApplication {
 
     public static void main(String[] args) {

iot-module/iot-module-data/iot-module-data-api/pom.xml

@@ -17,5 +17,14 @@
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
-
+    <dependencies>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-starter-feign</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-common</artifactId>
+        </dependency>
+    </dependencies>
 </project>

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/enums/ApiConstants.java

@@ -0,0 +1,12 @@
+package com.middle.platform.data.api.enums;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-03 10:28:58
+ */
+public class ApiConstants {
+    public static final String NAME = "iot-data";
+
+    public static final String VERSION = "1.0.0";
+
+}

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/feign/DynamicTopicApi.java

@@ -0,0 +1,21 @@
+package com.middle.platform.data.api.feign;
+
+import com.middle.platform.common.constant.RpcConstants;
+import com.middle.platform.data.api.enums.ApiConstants;
+import com.middle.platform.data.api.pojo.TopicDto;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-22 13:53:09
+ */
+@FeignClient(name = ApiConstants.NAME, contextId = "dynamicTopicApi")
+public interface DynamicTopicApi {
+    String PREFIX = RpcConstants.RPC_API_PREFIX + "/dynamic-topic";
+
+    @PostMapping(PREFIX + "/saveTopic")
+    Boolean saveTopic(@RequestBody TopicDto topicDto);
+
+}

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/pojo/TopicDto.java

@@ -0,0 +1,13 @@
+package com.middle.platform.data.api.pojo;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-17 16:05:28
+ */
+@Getter
+@Setter
+public class TopicDto {
+}

iot-module/iot-module-data/iot-module-data-biz/pom.xml

@@ -60,6 +60,10 @@
             <groupId>com.middle.platform</groupId>
             <artifactId>iot-starter-mqtt</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-module-data-api</artifactId>
+        </dependency>
 
     </dependencies>
 </project>

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/DataApplication.java

@@ -2,12 +2,14 @@ package com.middle.platform.data.biz;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 /**
  * @author xucaiqin
  * @date 2023-12-17 10:47:48
  */
 @SpringBootApplication
+@EnableTransactionManagement
 public class DataApplication {
 
     public static void main(String[] args) {

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/api/DynamicTopicApiImpl.java

@@ -0,0 +1,18 @@
+package com.middle.platform.data.biz.api;
+
+import com.middle.platform.data.api.feign.DynamicTopicApi;
+import com.middle.platform.data.biz.service.mqtt.impl.DynamicTopic;
+import lombok.RequiredArgsConstructor;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-22 13:55:46
+ */
+@RestController
+@Validated
+@RequiredArgsConstructor
+public class DynamicTopicApiImpl implements DynamicTopicApi {
+    private final DynamicTopic dynamicTopic;
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/constant/TopicConstant.java

@@ -0,0 +1,17 @@
+package com.middle.platform.data.biz.constant;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-22 11:44:53
+ */
+public interface TopicConstant {
+    //属性上报-发布
+    String ATTR_PUB = "1-1";
+    //云端响应属性上报-订阅
+    String ATTR_SUB = "1-2";
+    //事件上报
+    String EVENT_PUB = "2-1";
+    //云端响应事件上报-订阅
+    String EVENT_SUB = "2-2";
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/MqttService.java

@@ -2,6 +2,7 @@ package com.middle.platform.data.biz.service;
 
 import com.middle.platform.data.biz.config.MqttConfig;
 import com.middle.platform.data.biz.service.mqtt.MqttStrategyFactory;
+import com.middle.platform.data.biz.service.mqtt.MqttTopicStrategy;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Bean;
@@ -10,6 +11,8 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.stereotype.Component;
 
+import java.util.Objects;
+
 @Component
 @Slf4j
 public class MqttService {
@@ -23,10 +26,12 @@ public class MqttService {
             MessageHeaders headers = message.getHeaders();
             Object payload = message.getPayload();
             String topic = (String) headers.get("mqtt_receivedTopic");
-            log.info("topic {}",topic);
-            log.info("payload {}",payload);
-//            MqttTopicStrategy mqttTopicStrategy = mqttStrategyFactory.getByTopic(topic);
-//            mqttTopicStrategy.dealMsg(payload);
+            log.info("topic {}", topic);
+            log.info("payload {}", payload);
+            MqttTopicStrategy mqttTopicStrategy = mqttStrategyFactory.getByTopic(topic);
+            if (Objects.nonNull(mqttTopicStrategy)) {
+                mqttTopicStrategy.dealMsg(payload);
+            }
         };
     }
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/FuncPerEnum.java

@@ -0,0 +1,50 @@
+package com.middle.platform.data.biz.service.mqtt;
+
+import com.middle.platform.data.biz.constant.TopicConstant;
+import com.middle.platform.data.biz.service.mqtt.strategy.AttrReportStrategy;
+import com.middle.platform.data.biz.service.mqtt.strategy.EventReportStrategy;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * 功能权限 枚举
+ *
+ * @author xucaiqin
+ * @date 2023-12-22 11:37:56
+ */
+@Getter
+@AllArgsConstructor
+public enum FuncPerEnum {
+    ATTR_PUB(TopicConstant.ATTR_PUB, AttrReportStrategy.class.getSimpleName()),
+    EVENT_PUB(TopicConstant.EVENT_PUB, EventReportStrategy.class.getSimpleName()),
+    ;
+    private final String func;
+    private final String name;
+
+    /**
+     * 通过功能权限
+     *
+     * @param func 功能权限
+     *             <p>
+     *             属性上报字典-1
+     *             发布-1
+     *             订阅-2
+     *             发布订阅-3
+     *             如：属性上报-发布 具体为对应的值 1-1
+     *             </p>
+     * @return 策略类字符串
+     */
+    public static String strategy(String func) {
+        for (FuncPerEnum value : FuncPerEnum.values()) {
+            if (StringUtils.equals(value.getFunc(), func)) {
+                String beanName = value.getName();
+                char[] chars = beanName.toCharArray();
+                chars[0] += 32;
+                return String.valueOf(chars);
+            }
+        }
+        return null;
+    }
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/MqttStrategyFactory.java

@@ -9,9 +9,17 @@ import java.util.Map;
 public class MqttStrategyFactory {
     @Resource
     private Map<String, MqttTopicStrategy> mqttTopicStrategyMap;
+    @Resource
+    private StrategyAnalyze strategyAnalyze;
 
+    /**
+     * 通过topic名称获取对应解析策略
+     *
+     * @param topic mqtt主题
+     * @return
+     */
     public MqttTopicStrategy getByTopic(String topic) {
-        return mqttTopicStrategyMap.get(topic);
+        return mqttTopicStrategyMap.get(strategyAnalyze.getByTopic(topic));
     }
 
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/StrategyAnalyze.java

@@ -0,0 +1,38 @@
+package com.middle.platform.data.biz.service.mqtt;
+
+import com.middle.platform.redis.constant.CacheConstant;
+import jakarta.annotation.Resource;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * mqtt-topic解析
+ *
+ * @author xucaiqin
+ * @date 2023-12-22 11:27:55
+ */
+@Component
+public class StrategyAnalyze {
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
+
+    /**
+     * 设备上报数据，通过topic获取对应的策略类.
+     * redis:<topic,功能权限> 到 <功能权限,策略类>
+     *
+     * @param topic mqtt topic
+     * @return
+     */
+    public String getByTopic(String topic) {
+        //1.topic查找对应的功能权限
+        String func = (String) redisTemplate.opsForValue().get(String.format(CacheConstant.TOPIC_CACHE, topic));
+        if (StringUtils.isNotBlank(func)) {
+            //2.功能权限到策略类
+            return FuncPerEnum.strategy(func);
+        }
+        return "";
+    }
+
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/impl/DynamicTopic.java

@@ -0,0 +1,34 @@
+package com.middle.platform.data.biz.service.mqtt.impl;
+
+import jakarta.annotation.Resource;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-22 13:52:20
+ */
+@Component
+public class DynamicTopic {
+    @Resource
+    private MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter;
+
+    /**
+     * 云端动态订阅topic
+     *
+     * @param topic 订阅的topic
+     * @param qos   服务质量
+     */
+    public void addTopic(String topic, int qos) {
+        mqttPahoMessageDrivenChannelAdapter.addTopic(topic, qos);
+    }
+
+    /**
+     * 云端动态取消订阅topic
+     *
+     * @param topic 需要取消的topic
+     */
+    public void removeTopic(String topic) {
+        mqttPahoMessageDrivenChannelAdapter.removeTopic(topic);
+    }
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/impl/AttrReportStrategy.java → iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/strategy/AttrReportStrategy.java

@@ -1,4 +1,4 @@
-package com.middle.platform.data.biz.service.mqtt.impl;
+package com.middle.platform.data.biz.service.mqtt.strategy;
 
 import com.middle.platform.data.biz.service.mqtt.MqttTopicStrategy;
 import lombok.extern.slf4j.Slf4j;

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/strategy/EventReportStrategy.java

@@ -0,0 +1,19 @@
+package com.middle.platform.data.biz.service.mqtt.strategy;
+
+import com.middle.platform.data.biz.service.mqtt.MqttTopicStrategy;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * 事件上报
+ */
+@Service
+@Slf4j
+public class EventReportStrategy implements MqttTopicStrategy {
+
+    @Override
+    public void dealMsg(Object msg) {
+        log.debug("属性上报数据：{}", msg);
+
+    }
+}

iot-module/iot-module-manage/iot-module-manage-biz/pom.xml

@@ -66,5 +66,9 @@
             <groupId>com.middle.platform</groupId>
             <artifactId>iot-module-system-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-module-data-api</artifactId>
+        </dependency>
     </dependencies>
 </project>

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/config/FeignConfig.java

@@ -1,5 +1,6 @@
 package com.middle.platform.manage.biz.config;
 
+import com.middle.platform.data.api.feign.DynamicTopicApi;
 import com.middle.platform.system.api.feign.UserApi;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.Configuration;
@@ -11,6 +12,6 @@ import org.springframework.context.annotation.Configuration;
  * @date 2023-12-19 11:00:25
  */
 @Configuration
-@EnableFeignClients(clients = {UserApi.class})
+@EnableFeignClients(clients = {UserApi.class, DynamicTopicApi.class})
 public class FeignConfig {
 }

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/controller/IotProductController.java

@@ -137,7 +137,7 @@ public class IotProductController {
     @GetMapping("/getUrl")
     public Result<Object> getUrl(@RequestParam("productId") @NotBlank(message = "id不能为空") Long productId,
                                  @RequestParam("type") @NotNull(message = "类型不能为空") Integer type) {
-        return Result.ok(iotUrlService.getUrl(productId,type), "查询成功");
+        return Result.ok(iotUrlService.getUrl(productId, type), "查询成功");
     }
 
     /**
@@ -146,8 +146,8 @@ public class IotProductController {
      * @param iotUrlPara
      * @return
      */
-    @PostMapping("/urlConfig")
-    public Result<Object> urlConfig(@RequestBody @Valid IotUrlPara iotUrlPara) {
+    @PostMapping("/save")
+    public Result<Object> save(@RequestBody @Valid IotUrlPara iotUrlPara) {
         return Result.ok(iotUrlService.saveUrl(iotUrlPara), "保存成功");
     }
 

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/domain/req/IotProductPara.java

@@ -26,7 +26,7 @@ public class IotProductPara {
     private Integer nodeType;
 
     /**
-     * 联网类型
+     * 联网类型 network_type
      */
     @NotNull(message = "联网类型不能为空")
     private Integer networkType;

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/IotProductService.java

@@ -5,6 +5,7 @@ import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import com.middle.platform.common.exception.BusinessException;
 import com.middle.platform.common.utils.PageRes;
+import com.middle.platform.data.api.feign.DynamicTopicApi;
 import com.middle.platform.manage.biz.domain.req.IotProductPara;
 import com.middle.platform.manage.biz.domain.req.ProductPage;
 import com.middle.platform.manage.biz.domain.vo.IotDeviceDetailVo;
@@ -16,7 +17,6 @@ import com.middle.platform.manage.biz.mapper.IotProductMapper;
 import com.middle.platform.system.api.enums.DictType;
 import com.middle.platform.system.api.feign.DictApi;
 import com.middle.platform.system.api.feign.UserApi;
-import jakarta.annotation.Resource;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
@@ -38,10 +38,8 @@ public class IotProductService {
     private final IotModService iotModService;
     private final IotCloudService iotCloudService;
 
-    @Resource
-    private UserApi userApi;
-    @Resource
-    private  DictApi dictApi;
+    private final UserApi userApi;
+    private final DictApi dictApi;
 
     /**
      * 分页查询产品数据
@@ -73,6 +71,7 @@ public class IotProductService {
      * @param iotProductPara
      * @return
      */
+    @Transactional(rollbackFor = Exception.class)
     public Object save(IotProductPara iotProductPara) {
         IotProduct iotProduct = new IotProduct();
         iotProduct.setCode(IdUtil.nanoId(12));

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/IotUrlService.java

@@ -3,6 +3,7 @@ package com.middle.platform.manage.biz.service;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.middle.platform.common.constant.Global;
 import com.middle.platform.common.exception.BusinessException;
+import com.middle.platform.data.api.feign.DynamicTopicApi;
 import com.middle.platform.manage.biz.constant.UrlCategory;
 import com.middle.platform.manage.biz.constant.UrlInit;
 import com.middle.platform.manage.biz.constant.UrlProtocol;
@@ -23,6 +24,7 @@ import java.util.Objects;
 @RequiredArgsConstructor
 public class IotUrlService {
     private final IotUrlMapper iotUrlMapper;
+    private final DynamicTopicApi dynamicTopicApi;
 
     /**
      * 新增topic
@@ -43,6 +45,8 @@ public class IotUrlService {
         iotUrl.setType(iotUrlPara.getType());
         iotUrl.setRemark(iotUrlPara.getRemark());
         iotUrlMapper.insert(iotUrl);
+        //新增订阅的topic
+//        dynamicTopicApi
         return true;
     }