docs(数据流【设备网关-->物模型转换->时序入库】): 增加数据流入库的代码注释 优化代码结构

Signed-off-by: Tensai <517634644@qq.com>

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/MeasurementParameter.java

@@ -17,6 +17,7 @@ import java.util.Optional;
 public class MeasurementParameter implements ValueObject {
     private Map<String, Object> params = new HashMap<>();
 
+    @Override
     public Optional<Object> get(String name) {
         return Optional.ofNullable(params).map(p -> p.get(name));
     }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/DefaultElasticSearchService.java

@@ -37,7 +37,6 @@ import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
 import org.jetlinks.core.utils.FluxUtils;
 import org.reactivestreams.Publisher;
 import org.springframework.context.annotation.DependsOn;
-import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 import reactor.core.publisher.BufferOverflowStrategy;
 import reactor.core.publisher.Flux;
@@ -119,12 +118,14 @@ public class DefaultElasticSearchService implements ElasticSearchService {
             });
     }
 
+    @Override
     public <T> Flux<T> query(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
         return this
             .doQuery(new String[]{index}, queryParam)
             .flatMapMany(tp2 -> convertQueryResult(tp2.getT1(), tp2.getT2(), mapper));
     }
 
+    @Override
     public <T> Flux<T> query(String[] index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
         return this
             .doQuery(index, queryParam)
@@ -373,7 +374,7 @@ public class DefaultElasticSearchService implements ElasticSearchService {
 
     private Mono<SearchResponse> doSearch(SearchRequest request) {
         return this
-            .<SearchRequest, SearchResponse>execute(request, restClient.getQueryClient()::searchAsync)
+            .execute(request, restClient.getQueryClient()::searchAsync)
             .onErrorResume(err -> {
                 log.error("query elastic error", err);
                 return Mono.empty();

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/ElasticSearchService.java

@@ -12,6 +12,11 @@ import java.util.Collection;
 import java.util.Map;
 import java.util.function.Function;
 
+/**
+ * ES数据库业务操作类
+ *
+ * @author JetLinks
+ */
 public interface ElasticSearchService {
 
     default <T> Mono<PagerResult<T>> queryPager(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/reactive/ReactiveElasticSearchService.java

@@ -27,12 +27,12 @@ import org.hswebframework.utils.time.DateFormatter;
 import org.hswebframework.utils.time.DefaultDateFormatter;
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.bean.FastBeanCopier;
-import org.jetlinks.core.utils.FluxUtils;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.service.ElasticSearchService;
 import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
 import org.jetlinks.community.elastic.search.utils.QueryParamTranslator;
+import org.jetlinks.core.utils.FluxUtils;
 import org.reactivestreams.Publisher;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.DependsOn;
@@ -119,12 +119,14 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
             });
     }
 
+    @Override
     public <T> Flux<T> query(String index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
         return this
             .doQuery(new String[]{index}, queryParam)
             .flatMapMany(tp2 -> convertQueryResult(tp2.getT1(), tp2.getT2(), mapper));
     }
 
+    @Override
     public <T> Flux<T> query(String[] index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
         return this
             .doQuery(index, queryParam)

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/supports/DefaultDeviceGatewayManager.java

@@ -12,29 +12,53 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+/**
+ * 设备网关管理器
+ * <p>
+ * TCP   UDP   MQTT  CoAP
+ *
+ * @author Jetlinks
+ */
 @Component
 public class DefaultDeviceGatewayManager implements DeviceGatewayManager, BeanPostProcessor {
 
     private final DeviceGatewayPropertiesManager propertiesManager;
 
-    private Map<String, DeviceGatewayProvider> providers = new ConcurrentHashMap<>();
+    /**
+     * TCP MQTT的设备网关服务提供者
+     */
+    private final Map<String, DeviceGatewayProvider> providers = new ConcurrentHashMap<>();
 
-    private Map<String, DeviceGateway> store = new ConcurrentHashMap<>();
+    /**
+     * 启动状态的设备网关
+     */
+    private final Map<String, DeviceGateway> store = new ConcurrentHashMap<>();
 
     public DefaultDeviceGatewayManager(DeviceGatewayPropertiesManager propertiesManager) {
         this.propertiesManager = propertiesManager;
     }
 
+    /**
+     * 获取设备网关，有则返回，没有就创建返回
+     *
+     * @param id 网关ID
+     * @return 设备网关
+     */
     private Mono<DeviceGateway> doGetGateway(String id) {
         if (store.containsKey(id)) {
             return Mono.just(store.get(id));
         }
+
+        // 数据库查 DeviceGatewayEntity 转换成 DeviceGatewayProperties
+        // BeanMap中找provider 找不到就是不支持
+        // 创建设备网关
+        // double check 防止重复创建
         return propertiesManager
             .getProperties(id)
-            .switchIfEmpty(Mono.error(()->new UnsupportedOperationException("网关配置[" + id + "]不存在")))
+            .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("网关配置[" + id + "]不存在")))
             .flatMap(properties -> Mono
                 .justOrEmpty(providers.get(properties.getProvider()))
-                .switchIfEmpty(Mono.error(()->new UnsupportedOperationException("不支持的网络服务[" + properties.getProvider() + "]")))
+                .switchIfEmpty(Mono.error(() -> new UnsupportedOperationException("不支持的网络服务[" + properties.getProvider() + "]")))
                 .flatMap(provider -> provider
                     .createDeviceGateway(properties)
                     .flatMap(gateway -> {

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/supports/DeviceGatewayProperties.java

@@ -7,6 +7,9 @@ import org.jetlinks.community.ValueObject;
 import java.util.HashMap;
 import java.util.Map;
 
+/**
+ * @author Tensai
+ */
 @Getter
 @Setter
 public class DeviceGatewayProperties  implements ValueObject {

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/supports/DeviceGatewayPropertiesManager.java

@@ -2,8 +2,17 @@ package org.jetlinks.community.gateway.supports;
 
 import reactor.core.publisher.Mono;
 
+/**
+ * @author Jetlinks
+ */
 public interface DeviceGatewayPropertiesManager {
 
+    /**
+     * 获取网关的属性
+     *
+     * @param id 网关ID
+     * @return 网关属性
+     */
     Mono<DeviceGatewayProperties> getProperties(String id);
 
 

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/supports/DeviceGatewayProvider.java

@@ -4,6 +4,11 @@ import org.jetlinks.community.gateway.DeviceGateway;
 import org.jetlinks.community.network.NetworkType;
 import reactor.core.publisher.Mono;
 
+/**
+ * 设备网关服务提供者
+ *
+ * @author Jetlinks
+ */
 public interface DeviceGatewayProvider {
 
     String getId();

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/network/utils/DeviceGatewayHelper.java

@@ -21,6 +21,11 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+/**
+ * 设备网关处理工具
+ *
+ * @author Jetlinks
+ */
 @AllArgsConstructor
 public class DeviceGatewayHelper {
 
@@ -88,13 +93,13 @@ public class DeviceGatewayHelper {
                 return Mono
                     .delay(Duration.ofSeconds(2))
                     .then(registry
-                              .getDevice(children.getDeviceId())
-                              .flatMap(device -> device
-                                  //没有配置状态自管理才自动上线
-                                  .getSelfConfig(DeviceConfigKey.selfManageState)
-                                  .defaultIfEmpty(false)
-                                  .filter(Boolean.FALSE::equals)
-                                  .flatMap(ignore -> registerSession))
+                        .getDevice(children.getDeviceId())
+                        .flatMap(device -> device
+                            //没有配置状态自管理才自动上线
+                            .getSelfConfig(DeviceConfigKey.selfManageState)
+                            .defaultIfEmpty(false)
+                            .filter(Boolean.FALSE::equals)
+                            .flatMap(ignore -> registerSession))
                     );
             }
             return registerSession;
@@ -102,6 +107,15 @@ public class DeviceGatewayHelper {
         return Mono.empty();
     }
 
+    /**
+     * 处理来自设备网关的设备消息
+     *
+     * @param message                设备消息
+     * @param sessionBuilder         设备操作
+     * @param sessionConsumer        设备消费
+     * @param deviceNotFoundListener 异常监听
+     * @return 设备操作
+     */
     public Mono<DeviceOperator> handleDeviceMessage(DeviceMessage message,
                                                     Function<DeviceOperator, DeviceSession> sessionBuilder,
                                                     Consumer<DeviceSession> sessionConsumer,

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/device/TcpServerDeviceGateway.java

@@ -3,6 +3,16 @@ package org.jetlinks.community.network.tcp.device;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.logger.ReactiveLogger;
+import org.jetlinks.community.gateway.DeviceGateway;
+import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
+import org.jetlinks.community.gateway.monitor.GatewayMonitors;
+import org.jetlinks.community.gateway.monitor.MonitorSupportDeviceGateway;
+import org.jetlinks.community.network.DefaultNetworkType;
+import org.jetlinks.community.network.NetworkType;
+import org.jetlinks.community.network.tcp.TcpMessage;
+import org.jetlinks.community.network.tcp.client.TcpClient;
+import org.jetlinks.community.network.tcp.server.TcpServer;
+import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.device.DeviceOperator;
@@ -17,16 +27,6 @@ import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.DeviceGatewayContext;
 import org.jetlinks.core.server.session.DeviceSession;
 import org.jetlinks.core.server.session.DeviceSessionManager;
-import org.jetlinks.community.gateway.DeviceGateway;
-import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
-import org.jetlinks.community.gateway.monitor.GatewayMonitors;
-import org.jetlinks.community.gateway.monitor.MonitorSupportDeviceGateway;
-import org.jetlinks.community.network.DefaultNetworkType;
-import org.jetlinks.community.network.NetworkType;
-import org.jetlinks.community.network.tcp.TcpMessage;
-import org.jetlinks.community.network.tcp.client.TcpClient;
-import org.jetlinks.community.network.tcp.server.TcpServer;
-import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import reactor.core.Disposable;
 import reactor.core.publisher.EmitterProcessor;
@@ -43,11 +43,14 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.LongAdder;
 
 @Slf4j(topic = "system.tcp.gateway")
-class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGateway {
+public class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGateway {
 
     @Getter
     private final String id;
 
+    /**
+     * 维护所有创建的tcp server
+     */
     private final TcpServer tcpServer;
 
     private final String protocol;
@@ -60,6 +63,9 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
 
     private final DeviceGatewayMonitor gatewayMonitor;
 
+    /**
+     * 连接计数器
+     */
     private final LongAdder counter = new LongAdder();
 
     private final EmitterProcessor<Message> processor = EmitterProcessor.create(false);
@@ -67,10 +73,11 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
     private final FluxSink<Message> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
 
     private final AtomicBoolean started = new AtomicBoolean();
-
-    private Disposable disposable;
-
     private final DeviceGatewayHelper helper;
+    /**
+     * 数据流控开关
+     */
+    private Disposable disposable;
 
     public TcpServerDeviceGateway(String id,
                                   String protocol,
@@ -93,22 +100,88 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
         return supports.getProtocol(protocol);
     }
 
+    /**
+     * 当前总链接
+     *
+     * @return 当前总链接
+     */
     @Override
     public long totalConnection() {
         return counter.sum();
     }
 
+    /**
+     * 传输协议
+     *
+     * @return {@link org.jetlinks.core.message.codec.DefaultTransport}
+     */
     @Override
     public Transport getTransport() {
         return DefaultTransport.TCP;
     }
 
+    /**
+     * 网络类型
+     *
+     * @return {@link  org.jetlinks.community.network.DefaultNetworkType}
+     */
     @Override
     public NetworkType getNetworkType() {
         return DefaultNetworkType.TCP_SERVER;
     }
 
+    /**
+     * 启动网关
+     */
+    private void doStart() {
+        if (started.getAndSet(true) || disposable != null) {
+            return;
+        }
+        disposable = tcpServer
+            .handleConnection()
+            .publishOn(Schedulers.parallel())
+            .flatMap(client -> new TcpConnection(client).accept(), Integer.MAX_VALUE)
+            .onErrorContinue((err, obj) -> log.error(err.getMessage(), err))
+            .subscriberContext(ReactiveLogger.start("network", tcpServer.getId()))
+            .subscribe(
+                ignore -> {
+                },
+                error -> log.error(error.getMessage(), error)
+            );
+    }
+
+    @Override
+    public Flux<Message> onMessage() {
+        return processor;
+    }
+
+    @Override
+    public Mono<Void> pause() {
+        return Mono.fromRunnable(() -> started.set(false));
+    }
+
+    @Override
+    public Mono<Void> startup() {
+        return Mono.fromRunnable(this::doStart);
+    }
+
+    @Override
+    public Mono<Void> shutdown() {
+        return Mono.fromRunnable(() -> {
+            started.set(false);
+            disposable.dispose();
+            disposable = null;
+        });
+    }
+
+    @Override
+    public boolean isAlive() {
+        return started.get();
+    }
 
+    /**
+     * TCP 客户端连接
+     */
     class TcpConnection implements DeviceGatewayContext {
         final TcpClient client;
         final AtomicReference<Duration> keepaliveTimeout = new AtomicReference<>();
@@ -150,6 +223,11 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
 
         }
 
+        /**
+         * 接收消息
+         *
+         * @return null
+         */
         Mono<Void> accept() {
             return getProtocol()
                 .flatMap(protocol -> protocol.onClientConnect(getTransport(), client, this))
@@ -169,6 +247,12 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                 .doOnCancel(client::shutdown);
         }
 
+        /**
+         * 处理TCP消息 ==>> 设备消息
+         *
+         * @param message tcp消息
+         * @return null
+         */
         Mono<Void> handleTcpMessage(TcpMessage message) {
             return getProtocol()
                 .flatMap(pt -> pt.getMessageCodec(getTransport()))
@@ -177,28 +261,34 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                 .doOnNext(msg -> gatewayMonitor.receivedMessage())
                 .flatMap(this::handleDeviceMessage)
                 .doOnEach(ReactiveLogger.onError(err -> log.error("处理TCP[{}]消息失败:\n{}",
-                                                                  address,
-                                                                  message
+                    address,
+                    message
                     , err)))
                 .onErrorResume((err) -> Mono.fromRunnable(client::reset))
                 .then();
         }
 
+        /**
+         * 处理设备消息
+         *
+         * @param message 设备消息
+         * @return null
+         */
         Mono<Void> handleDeviceMessage(DeviceMessage message) {
             if (processor.hasDownstreams()) {
                 sink.next(message);
             }
             return helper
                 .handleDeviceMessage(message,
-                                     device -> new TcpDeviceSession(device, client, getTransport(), gatewayMonitor),
-                                     DeviceGatewayHelper
-                                         .applySessionKeepaliveTimeout(message, keepaliveTimeout::get)
-                                         .andThen(session -> {
-                                             TcpDeviceSession deviceSession = session.unwrap(TcpDeviceSession.class);
-                                             deviceSession.setClient(client);
-                                             sessionRef.set(deviceSession);
-                                         }),
-                                     () -> log.warn("无法从tcp[{}]消息中获取设备信息:{}", address, message)
+                    device -> new TcpDeviceSession(device, client, getTransport(), gatewayMonitor),
+                    DeviceGatewayHelper
+                        .applySessionKeepaliveTimeout(message, keepaliveTimeout::get)
+                        .andThen(session -> {
+                            TcpDeviceSession deviceSession = session.unwrap(TcpDeviceSession.class);
+                            deviceSession.setClient(client);
+                            sessionRef.set(deviceSession);
+                        }),
+                    () -> log.warn("无法从tcp[{}]消息中获取设备信息:{}", address, message)
                 )
                 .then();
         }
@@ -218,51 +308,4 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
             return handleDeviceMessage(message);
         }
     }
-
-
-    private void doStart() {
-        if (started.getAndSet(true) || disposable != null) {
-            return;
-        }
-        disposable = tcpServer
-            .handleConnection()
-            .publishOn(Schedulers.parallel())
-            .flatMap(client -> new TcpConnection(client).accept(), Integer.MAX_VALUE)
-            .onErrorContinue((err, obj) -> log.error(err.getMessage(), err))
-            .subscriberContext(ReactiveLogger.start("network", tcpServer.getId()))
-            .subscribe(
-                ignore -> {
-                },
-                error -> log.error(error.getMessage(), error)
-            );
-    }
-
-    @Override
-    public Flux<Message> onMessage() {
-        return processor;
-    }
-
-    @Override
-    public Mono<Void> pause() {
-        return Mono.fromRunnable(() -> started.set(false));
-    }
-
-    @Override
-    public Mono<Void> startup() {
-        return Mono.fromRunnable(this::doStart);
-    }
-
-    @Override
-    public Mono<Void> shutdown() {
-        return Mono.fromRunnable(() -> {
-            started.set(false);
-            disposable.dispose();
-            disposable = null;
-        });
-    }
-
-    @Override
-    public boolean isAlive() {
-        return started.get();
-    }
 }

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/server/TcpServer.java

@@ -23,5 +23,6 @@ public interface TcpServer extends Network {
     /**
      * 关闭服务端
      */
+    @Override
     void shutdown();
 }

jetlinks-components/timeseries-component/src/main/java/org/jetlinks/community/timeseries/TimeSeriesData.java

@@ -31,6 +31,7 @@ public interface TimeSeriesData extends ValueObject {
         return new SimpleTimeSeriesData(timestamp, data);
     }
 
+    @Override
     default <T> T as(Class<T> type) {
         return FastBeanCopier.copy(getData(), type);
     }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/message/DeviceMessageConnector.java

@@ -36,115 +36,9 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
         PropertyConstants.deviceName.getKey(),
         PropertyConstants.orgId.getKey()
     };
-
-    //设备注册中心
-    private final DeviceRegistry registry;
-
-    private final EventBus eventBus;
-
-    private final MessageHandler messageHandler;
-
     private final static BiConsumer<Throwable, Object> doOnError = (error, val) -> DeviceMessageConnector.log.error(error.getMessage(), error);
-
     private final static Function<DeviceOperator, Mono<Values>> configGetter = operator -> operator.getSelfConfigs(allConfigHeader);
-
     private final static Values emptyValues = Values.of(Collections.emptyMap());
-
-    public DeviceMessageConnector(EventBus eventBus,
-                                  DeviceRegistry registry,
-                                  MessageHandler messageHandler,
-                                  DeviceSessionManager sessionManager) {
-        this.registry = registry;
-        this.eventBus = eventBus;
-        this.messageHandler = messageHandler;
-        sessionManager
-            .onRegister()
-            .flatMap(session -> {
-                DeviceOnlineMessage message = new DeviceOnlineMessage();
-                message.setDeviceId(session.getDeviceId());
-                message.setTimestamp(session.connectTime());
-                return onMessage(message);
-            })
-            .onErrorContinue(doOnError)
-            .subscribe();
-
-        sessionManager
-            .onUnRegister()
-            .flatMap(session -> {
-                DeviceOfflineMessage message = new DeviceOfflineMessage();
-                message.setDeviceId(session.getDeviceId());
-                message.setTimestamp(System.currentTimeMillis());
-                return onMessage(message);
-            })
-            .onErrorContinue(doOnError)
-            .subscribe();
-    }
-
-    public Mono<Void> onMessage(Message message) {
-        if (null == message) {
-            return Mono.empty();
-        }
-        return this
-            .getTopic(message)
-            .flatMap(topic -> eventBus.publish(topic, message).then())
-            .onErrorContinue(doOnError)
-            .then();
-    }
-
-    private Flux<String> getTopic(Message message) {
-        Flux<String> topicsStream = createDeviceMessageTopic(registry, message);
-        if (message instanceof ChildDeviceMessage) { //子设备消息
-            return this
-                .onMessage(((ChildDeviceMessage) message).getChildDeviceMessage())
-                .thenMany(topicsStream);
-        } else if (message instanceof ChildDeviceMessageReply) { //子设备消息
-            return this
-                .onMessage(((ChildDeviceMessageReply) message).getChildDeviceMessage())
-                .thenMany(topicsStream);
-        }
-        return topicsStream;
-    }
-
-    public static Flux<String> createDeviceMessageTopic(DeviceRegistry deviceRegistry, Message message) {
-        return Flux.defer(() -> {
-            if (message instanceof DeviceMessage) {
-                DeviceMessage deviceMessage = ((DeviceMessage) message);
-                String deviceId = deviceMessage.getDeviceId();
-                if (deviceId == null) {
-                    log.warn("无法从消息中获取设备ID:{}", deviceMessage);
-                    return Mono.empty();
-                }
-                return deviceRegistry
-                    .getDevice(deviceId)
-                    .flatMap(configGetter)
-                    .defaultIfEmpty(emptyValues)
-                    .flatMapIterable(configs -> {
-                        configs.getAllValues().forEach(deviceMessage::addHeader);
-                        String productId = deviceMessage.getHeader(PropertyConstants.productId).orElse("null");
-                        String topic = createDeviceMessageTopic(productId, deviceId, deviceMessage);
-                        List<String> topics = new ArrayList<>(2);
-                        topics.add(topic);
-                        configs.getValue(PropertyConstants.orgId)
-                               .ifPresent(orgId -> topics.add("/org/" + orgId + topic));
-
-                        return topics;
-                    });
-            }
-            return Mono.just("/device/unknown/message/unknown");
-        });
-    }
-
-    public static String createDeviceMessageTopic(String productId, String deviceId, DeviceMessage message) {
-        StringBuilder builder = new StringBuilder(64)
-            .append("/device/")
-            .append(productId)
-            .append("/")
-            .append(deviceId);
-
-        appendDeviceMessageTopic(message, builder);
-        return builder.toString();
-    }
-
     private static final BiConsumer<Message, StringBuilder>[] fastTopicBuilder;
 
     static {
@@ -208,7 +102,7 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
             Message msg = ((ChildDeviceMessage) message).getChildDeviceMessage();
             if (msg instanceof DeviceMessage) {
                 builder.append("/message/children/")
-                       .append(((DeviceMessage) msg).getDeviceId());
+                    .append(((DeviceMessage) msg).getDeviceId());
             } else {
                 builder.append("/message/children");
             }
@@ -219,7 +113,7 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
             Message msg = ((ChildDeviceMessageReply) message).getChildDeviceMessage();
             if (msg instanceof DeviceMessage) {
                 builder.append("/message/children/reply/")
-                       .append(((DeviceMessage) msg).getDeviceId());
+                    .append(((DeviceMessage) msg).getDeviceId());
             } else {
                 builder.append("/message/children/reply");
             }
@@ -229,6 +123,81 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
         createFastBuilder(MessageType.DERIVED_METADATA, "/metadata/derived");
     }
 
+    //设备注册中心
+    private final DeviceRegistry registry;
+    private final EventBus eventBus;
+    private final MessageHandler messageHandler;
+
+    public DeviceMessageConnector(EventBus eventBus,
+                                  DeviceRegistry registry,
+                                  MessageHandler messageHandler,
+                                  DeviceSessionManager sessionManager) {
+        this.registry = registry;
+        this.eventBus = eventBus;
+        this.messageHandler = messageHandler;
+        sessionManager
+            .onRegister()
+            .flatMap(session -> {
+                DeviceOnlineMessage message = new DeviceOnlineMessage();
+                message.setDeviceId(session.getDeviceId());
+                message.setTimestamp(session.connectTime());
+                return onMessage(message);
+            })
+            .onErrorContinue(doOnError)
+            .subscribe();
+
+        sessionManager
+            .onUnRegister()
+            .flatMap(session -> {
+                DeviceOfflineMessage message = new DeviceOfflineMessage();
+                message.setDeviceId(session.getDeviceId());
+                message.setTimestamp(System.currentTimeMillis());
+                return onMessage(message);
+            })
+            .onErrorContinue(doOnError)
+            .subscribe();
+    }
+
+    public static Flux<String> createDeviceMessageTopic(DeviceRegistry deviceRegistry, Message message) {
+        return Flux.defer(() -> {
+            if (message instanceof DeviceMessage) {
+                DeviceMessage deviceMessage = ((DeviceMessage) message);
+                String deviceId = deviceMessage.getDeviceId();
+                if (deviceId == null) {
+                    log.warn("无法从消息中获取设备ID:{}", deviceMessage);
+                    return Mono.empty();
+                }
+                return deviceRegistry
+                    .getDevice(deviceId)
+                    .flatMap(configGetter)
+                    .defaultIfEmpty(emptyValues)
+                    .flatMapIterable(configs -> {
+                        configs.getAllValues().forEach(deviceMessage::addHeader);
+                        String productId = deviceMessage.getHeader(PropertyConstants.productId).orElse("null");
+                        String topic = createDeviceMessageTopic(productId, deviceId, deviceMessage);
+                        List<String> topics = new ArrayList<>(2);
+                        topics.add(topic);
+                        configs.getValue(PropertyConstants.orgId)
+                            .ifPresent(orgId -> topics.add("/org/" + orgId + topic));
+
+                        return topics;
+                    });
+            }
+            return Mono.just("/device/unknown/message/unknown");
+        });
+    }
+
+    public static String createDeviceMessageTopic(String productId, String deviceId, DeviceMessage message) {
+        StringBuilder builder = new StringBuilder(64)
+            .append("/device/")
+            .append(productId)
+            .append("/")
+            .append(deviceId);
+
+        appendDeviceMessageTopic(message, builder);
+        return builder.toString();
+    }
+
     private static void createFastBuilder(MessageType messageType,
                                           String topic) {
         fastTopicBuilder[messageType.ordinal()] = (ignore, builder) -> builder.append(topic);
@@ -249,6 +218,37 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
         }
     }
 
+    public Mono<Void> onMessage(Message message) {
+        if (null == message) {
+            return Mono.empty();
+        }
+        return this
+            .getTopic(message)
+            .flatMap(topic -> eventBus.publish(topic, message).then())
+            .onErrorContinue(doOnError)
+            .then();
+    }
+
+    private Flux<String> getTopic(Message message) {
+        Flux<String> topicsStream = createDeviceMessageTopic(registry, message);
+        if (message instanceof ChildDeviceMessage) { //子设备消息
+            return this
+                .onMessage(((ChildDeviceMessage) message).getChildDeviceMessage())
+                .thenMany(topicsStream);
+        } else if (message instanceof ChildDeviceMessageReply) { //子设备消息
+            return this
+                .onMessage(((ChildDeviceMessageReply) message).getChildDeviceMessage())
+                .thenMany(topicsStream);
+        }
+        return topicsStream;
+    }
+
+    /**
+     * 处理设备消息
+     *
+     * @param message 设备消息
+     * @return 处理结果
+     */
     protected Mono<Boolean> handleChildrenDeviceMessage(Message message) {
         if (message instanceof DeviceMessageReply) {
             return doReply(((DeviceMessageReply) message));
@@ -261,10 +261,23 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
         return handleChildrenDeviceMessage(reply.getChildDeviceMessage());
     }
 
+    /**
+     * 处理回复消息
+     *
+     * @param reply 子设备回复消息
+     * @return 处理结果
+     */
     protected Mono<Boolean> handleChildrenDeviceMessageReply(ChildDeviceMessageReply reply) {
         return handleChildrenDeviceMessage(reply.getChildDeviceMessage());
     }
 
+    /**
+     * 这里才是真正处理消息的地方
+     *
+     * @param device  设备操作类
+     * @param message 设备消息
+     * @return 处理结果
+     */
     @Override
     public Mono<Boolean> handleMessage(DeviceOperator device, @Nonnull Message message) {
         Mono<Boolean> then;
@@ -284,6 +297,12 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
 
     }
 
+    /**
+     * 回复消息处理逻辑
+     *
+     * @param reply 设备回复消息
+     * @return 处理结果
+     */
     private Mono<Boolean> doReply(DeviceMessageReply reply) {
         if (log.isDebugEnabled()) {
             log.debug("reply message {}", reply.getMessageId());

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/message/writer/TimeSeriesMessageWriterConnector.java

@@ -15,10 +15,16 @@ import reactor.core.publisher.Mono;
  */
 @Slf4j
 @AllArgsConstructor
-public class TimeSeriesMessageWriterConnector{
+public class TimeSeriesMessageWriterConnector {
     private final DeviceDataService dataService;
 
 
+    /**
+     * 订阅设备消息 入库
+     *
+     * @param message 设备消息
+     * @return null
+     */
     @Subscribe(topics = "/device/**", id = "device-message-ts-writer")
     public Mono<Void> writeDeviceMessageToTs(DeviceMessage message) {
         return dataService.saveDeviceMessage(message);

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/AbstractDeviceDataStoragePolicy.java

@@ -8,28 +8,22 @@ import org.hswebframework.ezorm.core.param.TermType;
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
 import org.hswebframework.web.id.IDGenerator;
+import org.jetlinks.community.device.entity.DeviceEvent;
+import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
+import org.jetlinks.community.device.entity.DeviceProperty;
+import org.jetlinks.community.device.enums.DeviceLogType;
+import org.jetlinks.community.device.events.handler.ValueTypeTranslator;
 import org.jetlinks.community.gateway.DeviceMessageUtils;
+import org.jetlinks.community.timeseries.TimeSeriesData;
 import org.jetlinks.core.device.DeviceConfigKey;
 import org.jetlinks.core.device.DeviceProductOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.DeviceLogMessage;
 import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.message.DeviceMessageReply;
 import org.jetlinks.core.message.Headers;
 import org.jetlinks.core.message.event.EventMessage;
-import org.jetlinks.core.message.property.ReadPropertyMessageReply;
-import org.jetlinks.core.message.property.ReportPropertyMessage;
-import org.jetlinks.core.message.property.WritePropertyMessageReply;
 import org.jetlinks.core.metadata.*;
 import org.jetlinks.core.metadata.types.*;
-import org.jetlinks.community.device.entity.DeviceEvent;
-import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
-import org.jetlinks.community.device.entity.DevicePropertiesEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.device.enums.DeviceLogType;
-import org.jetlinks.community.device.events.handler.ValueTypeTranslator;
-import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
-import org.jetlinks.community.timeseries.TimeSeriesData;
 import org.jetlinks.core.utils.DeviceMessageTracer;
 import org.jetlinks.core.utils.TimestampUtils;
 import org.reactivestreams.Publisher;
@@ -43,7 +37,6 @@ import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import static org.jetlinks.community.device.service.data.StorageConstants.propertyIsIgnoreStorage;
@@ -58,8 +51,8 @@ import static org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric.*;
  */
 public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
 
+    private final AtomicInteger nanoInc = new AtomicInteger();
     protected DeviceRegistry deviceRegistry;
-
     protected DeviceDataStorageProperties properties;
 
     public AbstractDeviceDataStoragePolicy(DeviceRegistry registry,
@@ -87,9 +80,11 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
     protected abstract Mono<Void> doSaveData(String metric, Flux<TimeSeriesData> data);
 
     /**
+     * 设备消息转换 二元组 {deviceId, tsData}
+     *
      * @param productId  产品ID
-     * @param message    原始消息
-     * @param properties 属性
+     * @param message    设备属性消息
+     * @param properties 物模型属性
      * @return 数据集合
      * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
      * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
@@ -106,7 +101,15 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
                                                              QueryParamEntity paramEntity,
                                                              Function<TimeSeriesData, T> mapper);
 
-
+    /**
+     * 保存单个设备消息,为了提升性能,存储策略会对保存请求进行缓冲,达到一定条件后
+     * 再进行批量写出,具体由不同对存储策略实现。
+     * <p>
+     * 如果保存失败,在这里不会得到错误信息.
+     *
+     * @param message 设备消息
+     * @return void
+     */
     @Nonnull
     @Override
     public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {
@@ -120,10 +123,10 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
     @Override
     public Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message) {
         return Flux.from(message)
-                   .flatMap(this::convertMessageToTimeSeriesData)
-                   .groupBy(Tuple2::getT1, Integer.MAX_VALUE)
-                   .flatMap(group -> doSaveData(group.key(), group.map(Tuple2::getT2)))
-                   .then();
+            .flatMap(this::convertMessageToTimeSeriesData)
+            .groupBy(Tuple2::getT1, Integer.MAX_VALUE)
+            .flatMap(group -> doSaveData(group.key(), group.map(Tuple2::getT2)))
+            .then();
     }
 
     protected String createDataId(DeviceMessage message) {
@@ -151,6 +154,12 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
             .toSimpleMap())));
     }
 
+    /**
+     * 设备消息转换成时序数据 二元组 {deviceId, tsData}
+     *
+     * @param message 设备消息
+     * @return 二元组
+     */
     protected Flux<Tuple2<String, TimeSeriesData>> convertMessageToTimeSeriesData(DeviceMessage message) {
         boolean ignoreStorage = message.getHeaderOrDefault(Headers.ignoreStorage);
         boolean ignoreLog = message.getHeaderOrDefault(Headers.ignoreLog);
@@ -193,8 +202,16 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
         return Flux.merge(all);
     }
 
+    /**
+     * 事件消息转换成 二元组{deviceId, tsData}
+     *
+     * @param productId 产品ID
+     * @param message   事件消息
+     * @return 二元组
+     */
     protected Mono<Tuple2<String, TimeSeriesData>> convertEventMessageToTimeSeriesData(String productId, EventMessage message) {
-
+        // 设备注册中心获取设备操作接口
+        // 获取设备元数据 物模型
         return deviceRegistry
             .getDevice(message.getDeviceId())
             .flatMap(device -> device
@@ -226,20 +243,19 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
             .map(data -> Tuples.of(deviceEventMetricId(productId, message.getEvent()), data));
     }
 
-
+    @Override
     public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId, @Nonnull QueryParamEntity entity) {
         return deviceRegistry
             .getDevice(deviceId)
             .flatMap(operator -> operator.getSelfConfig(DeviceConfigKey.productId))
             .flatMap(productId -> this
                 .doQueryPager(deviceLogMetricId(productId),
-                              entity.and("deviceId", TermType.eq, deviceId),
-                              data -> data.as(DeviceOperationLogEntity.class)
+                    entity.and("deviceId", TermType.eq, deviceId),
+                    data -> data.as(DeviceOperationLogEntity.class)
                 ))
             .defaultIfEmpty(PagerResult.empty());
     }
 
-
     @Nonnull
     @Override
     public Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
@@ -255,15 +271,15 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
                 .where("deviceId", deviceId)
                 .execute(param -> this
                     .doQuery(deviceEventMetricId(tp2.getT1().getId(), event),
-                             param,
-                             data -> {
-                                 DeviceEvent deviceEvent = new DeviceEvent(data.values());
-                                 if (format) {
-                                     deviceEvent.putFormat(tp2.getT2().getEventOrNull(event));
-                                 }
-                                 deviceEvent.putIfAbsent("timestamp", data.getTimestamp());
-                                 return deviceEvent;
-                             })));
+                        param,
+                        data -> {
+                            DeviceEvent deviceEvent = new DeviceEvent(data.values());
+                            if (format) {
+                                deviceEvent.putFormat(tp2.getT2().getEventOrNull(event));
+                            }
+                            deviceEvent.putIfAbsent("timestamp", data.getTimestamp());
+                            return deviceEvent;
+                        })));
     }
 
     @Nonnull
@@ -277,18 +293,18 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
             .getDevice(deviceId)
             .flatMap(device -> Mono.zip(device.getProduct(), device.getMetadata()))
             .flatMap(tp2 -> query.toQuery()
-                                 .where("deviceId", deviceId)
-                                 .execute(param -> this
-                                     .doQueryPager(deviceEventMetricId(tp2.getT1().getId(), event),
-                                                   param,
-                                                   data -> {
-                                                       DeviceEvent deviceEvent = new DeviceEvent(data.values());
-                                                       if (format) {
-                                                           deviceEvent.putFormat(tp2.getT2().getEventOrNull(event));
-                                                       }
-                                                       deviceEvent.putIfAbsent("timestamp", data.getTimestamp());
-                                                       return deviceEvent;
-                                                   }))
+                .where("deviceId", deviceId)
+                .execute(param -> this
+                    .doQueryPager(deviceEventMetricId(tp2.getT1().getId(), event),
+                        param,
+                        data -> {
+                            DeviceEvent deviceEvent = new DeviceEvent(data.values());
+                            if (format) {
+                                deviceEvent.putFormat(tp2.getT2().getEventOrNull(event));
+                            }
+                            deviceEvent.putIfAbsent("timestamp", data.getTimestamp());
+                            return deviceEvent;
+                        }))
             );
     }
 
@@ -382,6 +398,16 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
         return Maps.newHashMapWithExpectedSize(size);
     }
 
+    /**
+     * 设备消息转换 二元组{deviceId, tsData}
+     *
+     * @param productId  产品ID
+     * @param message    设备属性消息
+     * @param properties 物模型属性
+     * @return 数据集合
+     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
+     * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
+     */
     protected Flux<Tuple2<String, TimeSeriesData>> convertPropertiesForRowPolicy(String productId,
                                                                                  DeviceMessage message,
                                                                                  Map<String, Object> properties) {
@@ -414,10 +440,10 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
                         return Mono
                             .just(TimeSeriesData.of(ts, this
                                 .createRowPropertyData(id,
-                                                       TimestampUtils.toMillis(ts),
-                                                       device.getDeviceId(),
-                                                       propertyMetadata,
-                                                       entry.getT2().getValue()))
+                                    TimestampUtils.toMillis(ts),
+                                    device.getDeviceId(),
+                                    propertyMetadata,
+                                    entry.getT2().getValue()))
                             );
                     })
                     .map(data -> Tuples.of(devicePropertyMetricId(productId), data)))
@@ -511,9 +537,6 @@ public abstract class AbstractDeviceDataStoragePolicy implements DeviceDataStora
             .flatMap(product -> Mono.zip(Mono.just(product), product.getMetadata()));
     }
 
-
-    private final AtomicInteger nanoInc = new AtomicInteger();
-
     //将毫秒转为纳秒，努力让数据不重复
     protected long createUniqueNanoTime(long millis) {
         long nano = TimeUnit.MILLISECONDS.toNanos(millis);

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/DefaultDeviceDataService.java

@@ -2,16 +2,16 @@ package org.jetlinks.community.device.service.data;
 
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
+import org.jetlinks.community.device.entity.DeviceEvent;
+import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
+import org.jetlinks.community.device.entity.DeviceProperty;
+import org.jetlinks.community.timeseries.query.AggregationData;
 import org.jetlinks.core.Value;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceProductOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.DeviceMessage;
 import org.jetlinks.core.metadata.DeviceMetadata;
-import org.jetlinks.community.device.entity.DeviceEvent;
-import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.timeseries.query.AggregationData;
 import org.reactivestreams.Publisher;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.stereotype.Component;
@@ -23,6 +23,11 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 
+/**
+ * 默认设备数据服务
+ *
+ * @author JetLinks
+ */
 @Component
 public class DefaultDeviceDataService implements DeviceDataService {
 
@@ -55,8 +60,16 @@ public class DefaultDeviceDataService implements DeviceDataService {
             .then();
     }
 
+    /**
+     * 通过产品ID 获取存储策略
+     *
+     * @param productId 产品ID
+     * @return 存储策略
+     */
     Mono<DeviceDataStoragePolicy> getStoreStrategy(String productId) {
-
+        // 从注册中心获取产品操作接口
+        // 从配置中获取产品的存储策略
+        // 巧妙的双层switchIfEmpty 外层判断空配置 内层判断空策略
         return deviceRegistry
             .getProduct(productId)
             .flatMap(product -> product
@@ -69,7 +82,16 @@ public class DefaultDeviceDataService implements DeviceDataService {
                 .flatMap(Function.identity()));
     }
 
+    /**
+     * 通过设备ID 获取存储策略
+     *
+     * @param deviceId 设备ID
+     * @return 存储策略
+     */
     Mono<DeviceDataStoragePolicy> getDeviceStrategy(String deviceId) {
+        // 从注册中心获取设备操作接口
+        // 转换成产品操作接口
+        // 继而通过转换的产品ID获取存储策略
         return deviceRegistry.getDevice(deviceId)
             .flatMap(DeviceOperator::getProduct)
             .map(DeviceProductOperator::getId)
@@ -144,7 +166,15 @@ public class DefaultDeviceDataService implements DeviceDataService {
             .defaultIfEmpty(PagerResult.empty());
     }
 
-
+    /**
+     * 保存单个设备消息,为了提升性能,存储策略会对保存请求进行缓冲,达到一定条件后
+     * 再进行批量写出,具体由不同对存储策略实现。
+     * <p>
+     * 如果保存失败,在这里不会得到错误信息.
+     *
+     * @param message 设备消息
+     * @return void
+     */
     @Nonnull
     @Override
     public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/TimeSeriesColumnDeviceDataStoragePolicy.java

@@ -2,7 +2,14 @@ package org.jetlinks.community.device.service.data;
 
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.community.timeseries.query.*;
+import org.jetlinks.community.device.entity.DeviceProperty;
+import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetadata;
+import org.jetlinks.community.timeseries.TimeSeriesData;
+import org.jetlinks.community.timeseries.TimeSeriesManager;
+import org.jetlinks.community.timeseries.query.AggregationData;
+import org.jetlinks.community.timeseries.query.AggregationQueryParam;
+import org.jetlinks.community.timeseries.query.Group;
+import org.jetlinks.community.timeseries.query.TimeGroup;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.DeviceMessage;
@@ -10,10 +17,6 @@ import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.Converter;
 import org.jetlinks.core.metadata.DeviceMetadata;
 import org.jetlinks.core.metadata.PropertyMetadata;
-import org.jetlinks.community.device.entity.DeviceProperty;
-import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetadata;
-import org.jetlinks.community.timeseries.TimeSeriesData;
-import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.joda.time.DateTime;
 import org.joda.time.format.DateTimeFormat;
 import org.springframework.stereotype.Component;
@@ -28,7 +31,6 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric.devicePropertyMetric;
-import static org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric.devicePropertyMetricId;
 
 @Component
 public class TimeSeriesColumnDeviceDataStoragePolicy extends TimeSeriesDeviceDataStoragePolicy implements DeviceDataStoragePolicy {
@@ -247,11 +249,22 @@ public class TimeSeriesColumnDeviceDataStoragePolicy extends TimeSeriesDeviceDat
             .doOnNext(agg -> agg.values().remove("_time"));
     }
 
+    /**
+     * 设备消息转换 二元组{deviceId, tsData}
+     *
+     * @param productId  产品ID
+     * @param message    设备属性消息
+     * @param properties 物模型属性
+     * @return 数据集合
+     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
+     * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
+     */
     @Override
     protected Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId, DeviceMessage message, Map<String, Object> properties) {
         return convertPropertiesForColumnPolicy(productId, message, properties);
     }
 
+    @Override
     protected Object convertPropertyValue(Object value, PropertyMetadata metadata) {
         if (value == null || metadata == null) {
             return value;

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/TimeSeriesDeviceDataStoragePolicy.java

@@ -22,18 +22,21 @@ public abstract class TimeSeriesDeviceDataStoragePolicy extends AbstractDeviceDa
         this.timeSeriesManager = timeSeriesManager;
     }
 
+    @Override
     protected Mono<Void> doSaveData(String metric, TimeSeriesData data) {
         return timeSeriesManager
             .getService(metric)
             .commit(data);
     }
 
+    @Override
     protected Mono<Void> doSaveData(String metric, Flux<TimeSeriesData> data) {
         return timeSeriesManager
             .getService(metric)
             .save(data);
     }
 
+    @Override
     protected <T> Flux<T> doQuery(String metric,
                                   QueryParamEntity paramEntity,
                                   Function<TimeSeriesData, T> mapper) {
@@ -44,6 +47,7 @@ public abstract class TimeSeriesDeviceDataStoragePolicy extends AbstractDeviceDa
     }
 
 
+    @Override
     protected <T> Mono<PagerResult<T>> doQueryPager(String metric,
                                                     QueryParamEntity paramEntity,
                                                     Function<TimeSeriesData, T> mapper) {

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/data/TimeSeriesRowDeviceDataStoreStoragePolicy.java

@@ -2,18 +2,18 @@ package org.jetlinks.community.device.service.data;
 
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
-import org.jetlinks.core.device.DeviceOperator;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.metadata.ConfigMetadata;
-import org.jetlinks.core.metadata.DeviceMetadata;
-import org.jetlinks.core.metadata.PropertyMetadata;
 import org.jetlinks.community.device.entity.DeviceProperty;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetadata;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.timeseries.TimeSeriesData;
 import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.community.timeseries.query.*;
+import org.jetlinks.core.device.DeviceOperator;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.core.message.DeviceMessage;
+import org.jetlinks.core.metadata.ConfigMetadata;
+import org.jetlinks.core.metadata.DeviceMetadata;
+import org.jetlinks.core.metadata.PropertyMetadata;
 import org.jetlinks.reactor.ql.utils.CastUtils;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
@@ -288,6 +288,16 @@ public class TimeSeriesRowDeviceDataStoreStoragePolicy extends TimeSeriesDeviceD
             .doOnNext(agg -> agg.values().remove("_time"));
     }
 
+    /**
+     * 设备消息转换 二元组{deviceId, tsData}
+     *
+     * @param productId  产品ID
+     * @param message    设备属性消息
+     * @param properties 物模型属性
+     * @return 数据集合
+     * @see this#convertPropertiesForColumnPolicy(String, DeviceMessage, Map)
+     * @see this#convertPropertiesForRowPolicy(String, DeviceMessage, Map)
+     */
     @Override
     protected Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId, DeviceMessage message, Map<String, Object> properties) {
         return convertPropertiesForRowPolicy(productId, message, properties);

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/timeseries/DeviceTimeSeriesMetric.java

@@ -1,8 +1,8 @@
 package org.jetlinks.community.device.timeseries;
 
+import org.jetlinks.community.timeseries.TimeSeriesMetric;
 import org.jetlinks.core.device.DeviceProductOperator;
 import org.jetlinks.core.metadata.EventMetadata;
-import org.jetlinks.community.timeseries.TimeSeriesMetric;
 
 /**
  * 设备时序数据度量标识
@@ -26,6 +26,13 @@ public interface DeviceTimeSeriesMetric {
         return TimeSeriesMetric.of(deviceEventMetricId(productId, eventId));
     }
 
+    /**
+     * 构建事件指标ID
+     *
+     * @param productId 产品ID
+     * @param eventId   事件ID
+     * @return 事件指标ID
+     */
     static String deviceEventMetricId(String productId, String eventId) {
         return "event_".concat(productId).concat("_").concat(eventId);
     }

jetlinks-manager/network-manager/src/main/java/org/jetlinks/community/network/manager/service/DeviceGatewayConfigService.java

@@ -7,6 +7,9 @@ import org.jetlinks.community.gateway.supports.DeviceGatewayPropertiesManager;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
+/**
+ * @author jetlinks
+ */
 @Service
 public class DeviceGatewayConfigService implements DeviceGatewayPropertiesManager {