reactor：【IoT 物联网】数据下行，基于 messagebus 实现订阅消费

YunaiV
1 parent 1b59aa9c
Showing 7 changed files with 64 additions and 55 deletions
yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/json/JsonUtils.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceController.http
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/control/IotDeviceDownstreamServiceImpl.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java
yudao-module-iot/yudao-module-iot-net-components/yudao-module-iot-net-component-http/src/main/java/cn/iocoder/yudao/module/iot/net/component/http/config/IotNetComponentHttpAutoConfiguration.java
yudao-module-iot/yudao-module-iot-net-components/yudao-module-iot-net-component-http/src/main/java/cn/iocoder/yudao/module/iot/net/component/http/upstream/IotDeviceUpstreamServer.java
@@ -3,18 +3,22 @@ package cn.iocoder.yudao.framework.common.util.json;
 import cn.hutool.core.util.ArrayUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONUtil;
+import cn.iocoder.yudao.framework.common.util.json.databind.TimestampLocalDateTimeDeserializer;
+import cn.iocoder.yudao.framework.common.util.json.databind.TimestampLocalDateTimeSerializer;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import java.io.IOException;
 import java.lang.reflect.Type;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
@@ -32,7 +36,11 @@ public class JsonUtils {
         objectMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
         objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL); // 忽略 null 值
-        objectMapper.registerModules(new JavaTimeModule()); // 解决 LocalDateTime 的序列化
+        // 解决 LocalDateTime 的序列化
+        SimpleModule simpleModule = new JavaTimeModule()
+                .addSerializer(LocalDateTime.class, TimestampLocalDateTimeSerializer.INSTANCE)
+                .addDeserializer(LocalDateTime.class, TimestampLocalDateTimeDeserializer.INSTANCE);
+        objectMapper.registerModules(simpleModule);
     }
     /**
@@ -16,7 +16,7 @@ Authorization: Bearer {{token}}
 ### 请求 /iot/device/downstream 接口（属性设置） => 成功
 POST {{baseUrl}}/iot/device/downstream
 Content-Type: application/json
-tenant-id: {{adminTenentId}}
+tenant-id: {{adminTenantId}}
 Authorization: Bearer {{token}}
 {
@@ -8,6 +8,7 @@ import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.module.iot.api.device.dto.control.downstream.*;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceDownstreamReqVO;
+import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.plugin.IotPluginInstanceDO;
 import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageIdentifierEnum;
@@ -52,6 +53,8 @@ public class IotDeviceDownstreamServiceImpl implements IotDeviceDownstreamServic
     @Resource
     private IotDeviceProducer deviceProducer;
+    @Resource
+    private IotDeviceMessageProducer deviceMessageProducer;
     @Override
     public IotDeviceMessage downstreamDevice(IotDeviceDownstreamReqVO downstreamReqVO) {
@@ -150,26 +153,16 @@ public class IotDeviceDownstreamServiceImpl implements IotDeviceDownstreamServic
         // TODO @super：【可优化】过滤掉不合法的属性
         // 2. 发送请求
-        String url = String.format("sys/%s/%s/thing/service/property/set",
-                getProductKey(device, parentDevice), getDeviceName(device, parentDevice));
-        IotDevicePropertySetReqDTO reqDTO = new IotDevicePropertySetReqDTO()
-                .setProperties((Map<String, Object>) downstreamReqVO.getData());
-        CommonResult<Boolean> result = requestPlugin(url, reqDTO, device);
-
-        // 3. 发送设备消息
-        IotDeviceMessage message = new IotDeviceMessage().setRequestId(reqDTO.getRequestId())
-                .setType(IotDeviceMessageTypeEnum.PROPERTY.getType())
-                .setIdentifier(IotDeviceMessageIdentifierEnum.PROPERTY_SET.getIdentifier())
-                .setData(reqDTO.getProperties());
-        sendDeviceMessage(message, device, result.getCode());
-
-        // 4. 如果不成功，抛出异常，提示用户
-        if (result.isError()) {
-            log.error("[setDeviceProperty][设备({})属性设置失败，请求参数：({})，响应结果：({})]",
-                    device.getDeviceKey(), reqDTO, result);
-            throw exception(DEVICE_DOWNSTREAM_FAILED, result.getMsg());
-        }
-        return message;
+        // TODO @芋艿：deviceName 的设置
+        String deviceName = "xx";
+        Long tenantId = 1L;
+        cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage message = cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage
+                .of(getProductKey(device, parentDevice), getDeviceName(device, parentDevice), deviceName,
+                        null, tenantId);
+        String serverId = "yy";
+        deviceMessageProducer.sendGatewayDeviceMessage(serverId, message);
+        // TODO @芋艿：后续可以清理掉
+        return null;
     }
     /**
@@ -32,7 +32,7 @@ public class IotDeviceMessage {
      *
      * 其中，%s 就是该“server”(protocol) 的标识
      */
-    public static final String MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC = MESSAGE_BUS_DEVICE_MESSAGE_TOPIC + "/%s";
+    public static final String MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC = MESSAGE_BUS_DEVICE_MESSAGE_TOPIC + "_%s";
     /**
      * 请求编号
@@ -71,6 +71,7 @@ public class IotDeviceMessage {
      * 例如说：属性上报的 properties、事件上报的 params
      */
     private Object data;
+    // TODO @芋艿：可能会去掉
     /**
      * 响应码
      *
@@ -100,6 +101,20 @@ public class IotDeviceMessage {
         return this;
     }
+    public IotDeviceMessage ofPropertySet(Map<String, Object> properties) {
+        this.setType(IotDeviceMessageTypeEnum.PROPERTY.getType());
+        this.setIdentifier(IotDeviceMessageIdentifierEnum.PROPERTY_SET.getIdentifier());
+        this.setData(properties);
+        return this;
+    }
+
+    public static IotDeviceMessage of(String productKey, String deviceName, String deviceKey,
+                                      String serverId, Long tenantId) {
+        return of(productKey, deviceName, deviceKey,
+                null, null,
+                serverId, tenantId);
+    }
+
     public static IotDeviceMessage of(String productKey, String deviceName, String deviceKey,
                                       String requestId, LocalDateTime reportTime,
                                       String serverId, Long tenantId) {
@@ -117,7 +132,7 @@ public class IotDeviceMessage {
     // ========== Topic 相关 ==========
-    public static String getMessageBusGatewayDeviceMessageTopic(String serverId) {
+    public static String buildMessageBusGatewayDeviceMessageTopic(String serverId) {
         return String.format(MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC, serverId);
     }
@@ -30,7 +30,7 @@ public class IotDeviceMessageProducer {
      * @param message 设备消息
      */
     public void sendGatewayDeviceMessage(String serverId, Object message) {
-        messageBus.post(IotDeviceMessage.getMessageBusGatewayDeviceMessageTopic(serverId), message);
+        messageBus.post(IotDeviceMessage.buildMessageBusGatewayDeviceMessageTopic(serverId), message);
     }
 }
 package cn.iocoder.yudao.module.iot.net.component.http.config;
-import cn.hutool.system.SystemUtil;
 import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBusSubscriber;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
-import cn.iocoder.yudao.module.iot.net.component.core.config.IotNetComponentCommonProperties;
 import cn.iocoder.yudao.module.iot.net.component.core.downstream.IotDeviceDownstreamHandler;
-import cn.iocoder.yudao.module.iot.net.component.core.heartbeat.IotNetComponentRegistry;
-import cn.iocoder.yudao.module.iot.net.component.core.util.IotNetComponentCommonUtils;
 import cn.iocoder.yudao.module.iot.net.component.http.downstream.IotDeviceDownstreamHandlerImpl;
 import cn.iocoder.yudao.module.iot.net.component.http.upstream.IotDeviceUpstreamServer;
 import io.vertx.core.Vertx;
@@ -36,15 +35,6 @@ import org.springframework.context.event.EventListener;
 public class IotNetComponentHttpAutoConfiguration {
     /**
-     * 组件 key
-     */
-    private static final String PLUGIN_KEY = "http";
-
-    public IotNetComponentHttpAutoConfiguration() {
-        // 构造函数中不输出日志，移到 initialize 方法中
-    }
-
-    /**
      * 初始化 HTTP 组件
      *
      * @param event 应用启动事件
@@ -53,27 +43,30 @@ public class IotNetComponentHttpAutoConfiguration {
     public void initialize(ApplicationStartedEvent event) {
         log.info("[IotNetComponentHttpAutoConfiguration][开始初始化]");
-        // 从应用上下文中获取需要的 Bean
-        IotNetComponentRegistry componentRegistry = event.getApplicationContext()
-                .getBean(IotNetComponentRegistry.class);
-        IotNetComponentCommonProperties commonProperties = event.getApplicationContext()
-                .getBean(IotNetComponentCommonProperties.class);
+        // TODO @芋艿：临时处理
+        IotMessageBus messageBus = event.getApplicationContext()
+                .getBean(IotMessageBus.class);
+        messageBus.register(new IotMessageBusSubscriber<IotDeviceMessage>() {
+
+            @Override
+            public String getTopic() {
+                return IotDeviceMessage.buildMessageBusGatewayDeviceMessageTopic("yy");
+            }
-        // 设置当前组件的核心标识
-        // 注意：这里只为当前 HTTP 组件设置 pluginKey，不影响其他组件
-        // TODO @haohao：多个会存在冲突的问题哇？
-        commonProperties.setPluginKey(PLUGIN_KEY);
+            @Override
+            public String getGroup() {
+                return "test";
+            }
-        // 将 HTTP 组件注册到组件注册表
-        componentRegistry.registerComponent(
-                PLUGIN_KEY,
-                SystemUtil.getHostInfo().getAddress(),
-                0, // 内嵌模式固定为 0：自动生成对应的 port 端口号
-                IotNetComponentCommonUtils.getProcessId());
+            @Override
+            public void onMessage(IotDeviceMessage message) {
+                System.out.println(message);
+            }
-        log.info("[initialize][IoT HTTP 组件初始化完成]");
+        });
     }
+    // TODO @芋艿：貌似这里不用注册 bean？
     /**
      * 创建 Vert.x 实例
      *
@@ -32,7 +32,7 @@ public class IotDeviceUpstreamServer extends AbstractVerticle {
     private final IotDeviceMessageProducer deviceMessageProducer;
     @Override
-    public void start() throws Exception {
+    public void start() {
         start(Promise.promise());
     }