Merge remote-tracking branch 'origin/master'

# Conflicts:
#	jetlinks-manager/authentication-manager/src/main/java/org/jetlinks/community/auth/web/OrganizationController.java

.github/workflows/pull_request.yml

@@ -0,0 +1,27 @@
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+
+name: Java CI with Maven
+
+on:
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v1
+    - name: Set up JDK 1.8
+      uses: actions/setup-java@v1
+      with:
+        java-version: 1.8
+    - name: Cache Maven Repository
+      uses: actions/cache@v1
+      with:
+        path: ~/.m2
+        key: jetlinks-community-maven-repository
+    - name: Build with Maven
+      run: ./mvnw package -Dmaven.test.skip=true -Pbuild

.gitignore

@@ -26,4 +26,5 @@ hs_err_pid*
 /ui/upload/
 docker/data
 !device-simulator.jar
-!demo-protocol-1.0.jar
+!demo-protocol-1.0.jar
+application-local.yml

.mvn/wrapper/maven-wrapper.properties

@@ -1 +1 @@
-distributionUrl=http://mirrors.hust.edu.cn/apache/maven/maven-3/3.3.9/binaries/apache-maven-3.3.9-bin.zip
+distributionUrl=https://downloads.apache.org/maven/maven-3/3.3.9/binaries/apache-maven-3.3.9-bin.zip

README.md

@@ -1,8 +1,10 @@
 # JetLinks 物联网基础平台
 
 ![GitHub Workflow Status](https://img.shields.io/github/workflow/status/jetlinks/jetlinks-community/Auto%20Deploy%20Docker?label=docker)
-![Version](https://img.shields.io/badge/Version-1.8--RELEASE-brightgreen)
-![QQ群2021514](https://img.shields.io/badge/QQ群-2021514-brightgreen)
+![Version](https://img.shields.io/badge/version-1.9--RELEASE-brightgreen)
+[![Codacy Badge](https://api.codacy.com/project/badge/Grade/e8d527d692c24633aba4f869c1c5d6ad)](https://app.codacy.com/gh/jetlinks/jetlinks-community?utm_source=github.com&utm_medium=referral&utm_content=jetlinks/jetlinks-community&utm_campaign=Badge_Grade_Settings)
+[![QQ①群2021514](https://img.shields.io/badge/QQ①群-2021514-brightgreen)](https://qm.qq.com/cgi-bin/qm/qr?k=LGf0OPQqvLGdJIZST3VTcypdVWhdfAOG&jump_from=webapi)
+[![QQ②群324606263](https://img.shields.io/badge/QQ②群-324606263-brightgreen)](https://qm.qq.com/cgi-bin/qm/qr?k=IMas2cH-TNsYxUcY8lRbsXqPnA2sGHYQ&jump_from=webapi)
 ![jetlinks](https://visitor-badge.glitch.me/badge?page_id=jetlinks)
 
 JetLinks 基于Java8,Spring Boot 2.x,WebFlux,Netty,Vert.x,Reactor等开发, 
@@ -22,7 +24,8 @@ JetLinks 基于Java8,Spring Boot 2.x,WebFlux,Netty,Vert.x,Reactor等开发,
 
 地理位置:统一管理地理位置信息,支持区域搜索. 
 
-官方QQ群: `2021514`
+官方QQ: ①群 [2021514](https://qm.qq.com/cgi-bin/qm/qr?k=LGf0OPQqvLGdJIZST3VTcypdVWhdfAOG&jump_from=webapi)
+, ②群 [324606263](https://qm.qq.com/cgi-bin/qm/qr?k=IMas2cH-TNsYxUcY8lRbsXqPnA2sGHYQ&jump_from=webapi)
 
 ## 技术栈
 

docker/run-all/docker-compose-embedded.yml

@@ -1,51 +0,0 @@
-version: '2'
-services:
-  redis:
-    image: redis:5.0.4
-    container_name: jetlinks-ce-redis
-    #    ports:
-    #      - "6379:6379"
-    volumes:
-      - "redis-volume:/data"
-    command: redis-server --appendonly yes
-    environment:
-      - TZ=Asia/Shanghai
-  ui:
-    image: registry.cn-shenzhen.aliyuncs.com/jetlinks/jetlinks-ui-antd:1.4.0
-    container_name: jetlinks-ce-ui
-    ports:
-      - 9000:80
-    environment:
-      - "API_BASE_PATH=http://jetlinks:8848/" #API根路径
-    volumes:
-      - "jetlinks-upload-volume:/usr/share/nginx/html/upload"
-    links:
-      - jetlinks:jetlinks
-  jetlinks:
-    image: registry.cn-shenzhen.aliyuncs.com/jetlinks/jetlinks-standalone:1.4.0
-    container_name: jetlinks-ce
-    ports:
-      - 8848:8848 # API端口
-      - 1883:1883 # MQTT端口
-      - 8000:8000 # 预留
-      - 8001:8001 # 预留
-      - 8002:8002 # 预留
-      - 9000:9000 # elasticsearch
-      - 6379:6379 # redis
-    volumes:
-      - "jetlinks-upload-volume:/static/upload"  # 持久化上传的文件
-      - "jetlinks-data-volume:/data"
-    environment:
-     # - "JAVA_OPTS=-Xms4g -Xmx18g -XX:+UseG1GC"
-      - "spring.profiles.active=dev,embedded" #使用dev和embedded环境.
-      - "hsweb.file.upload.static-location=http://127.0.0.1:8848/upload"  #上传的静态文件访问根地址,为ui的地址.
-      - "logging.level.io.r2dbc=warn"
-      - "spring.redis.host=redis"
-      - "logging.level.org.springframework.data=warn"
-      - "logging.level.org.springframework=warn"
-      - "logging.level.org.jetlinks=warn"
-      - "logging.level.org.hswebframework=warn"
-      - "logging.level.org.springframework.data.r2dbc.connectionfactory=warn"
-volumes:
-  jetlinks-upload-volume:
-  jetlinks-data-volume:

docker/run-all/docker-compose.yml

@@ -7,7 +7,7 @@ services:
     #      - "6379:6379"
     volumes:
       - "redis-volume:/data"
-    command: redis-server --appendonly yes
+    command: redis-server --appendonly yes --requirepass "JetLinks@redis"
     environment:
       - TZ=Asia/Shanghai
   elasticsearch:
@@ -48,7 +48,7 @@ services:
       POSTGRES_DB: jetlinks
       TZ: Asia/Shanghai
   ui:
-    image: registry.cn-shenzhen.aliyuncs.com/jetlinks/jetlinks-ui-antd:1.9.0
+    image: registry.cn-shenzhen.aliyuncs.com/jetlinks/jetlinks-ui-antd:1.10.0
     container_name: jetlinks-ce-ui
     ports:
       - 9000:80
@@ -59,26 +59,29 @@ services:
     links:
       - jetlinks:jetlinks
   jetlinks:
-    image: registry.cn-shenzhen.aliyuncs.com/jetlinks/jetlinks-standalone:1.9.0-SNAPSHOT
+    image: registry.cn-shenzhen.aliyuncs.com/jetlinks/jetlinks-standalone:1.10.0
     container_name: jetlinks-ce
     ports:
-      - 8848:8848 # API端口
-      - 1883-1890:1883-1890 # 预留
-      - 8000-8010:8000-8010 # 预留
+      - "8848:8848" # API端口
+      - "1883-1890:1883-1890" # 预留
+      - "8000-8010:8000-8010" # 预留
     volumes:
       - "jetlinks-volume:/application/static/upload"  # 持久化上传的文件
       - "jetlinks-protocol-volume:/application/data/protocols"
     environment:
-     # - "JAVA_OPTS=-Xms4g -Xmx18g -XX:+UseG1GC"
+      - "JAVA_OPTS=-Duser.language=zh -XX:+UseG1GC"
       - "TZ=Asia/Shanghai"
       - "hsweb.file.upload.static-location=http://127.0.0.1:8848/upload"  #上传的静态文件访问根地址,为ui的地址.
       - "spring.r2dbc.url=r2dbc:postgresql://postgres:5432/jetlinks" #数据库连接地址
       - "spring.r2dbc.username=postgres"
       - "spring.r2dbc.password=jetlinks"
-      - "elasticsearch.client.host=elasticsearch"
-      - "elasticsearch.client.post=9200"
+      - "spring.data.elasticsearch.client.reactive.endpoints=elasticsearch:9200"
+#        - "spring.data.elasticsearch.client.reactive.username=admin"
+#        - "spring.data.elasticsearch.client.reactive.password=admin"
+#        - "spring.reactor.debug-agent.enabled=false" #设置为false能提升性能
       - "spring.redis.host=redis"
       - "spring.redis.port=6379"
+      - "spring.redis.password=JetLinks@redis"
       - "logging.level.io.r2dbc=warn"
       - "logging.level.org.springframework.data=warn"
       - "logging.level.org.springframework=warn"

jetlinks-components/common-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -34,9 +34,15 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-core</artifactId>
         </dependency>
+
         <dependency>
             <groupId>org.jetlinks</groupId>
             <artifactId>reactor-ql</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webflux</artifactId>
+        </dependency>
     </dependencies>
 </project>

jetlinks-components/common-component/src/main/java/org/jetlinks/community/ConfigMetadataConstants.java

@@ -3,6 +3,9 @@ package org.jetlinks.community;
 import org.jetlinks.core.config.ConfigKey;
 
 /**
+ * 数据验证配置常量类
+ *
+ * @author zhouhao
  * @see ConfigKey
  */
 public interface ConfigMetadataConstants {

jetlinks-components/common-component/src/main/java/org/jetlinks/community/PropertyConstants.java

@@ -3,7 +3,6 @@ package org.jetlinks.community;
 import org.jetlinks.core.config.ConfigKey;
 import org.jetlinks.core.message.HeaderKey;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 

jetlinks-components/common-component/src/main/java/org/jetlinks/community/PropertyMetadataConstants.java

@@ -0,0 +1,85 @@
+package org.jetlinks.community;
+
+import org.jetlinks.core.metadata.PropertyMetadata;
+import org.jetlinks.reactor.ql.utils.CastUtils;
+
+public interface PropertyMetadataConstants {
+
+    /**
+     * 属性来源
+     */
+    interface Source {
+        //数据来源
+        String id = "source";
+
+        //手动写值
+        String manual = "manual";
+
+        //规则,虚拟属性
+        String rule = "rule";
+
+        /**
+         * 判断属性是否手动赋值
+         *
+         * @param metadata 属性物模型
+         * @return 是否手动赋值
+         */
+        static boolean isManual(PropertyMetadata metadata) {
+            return metadata.getExpand(id)
+                           .map(manual::equals)
+                           .orElse(false);
+        }
+
+        /**
+         * 判断属性是否为规则
+         *
+         * @param metadata 物模型
+         * @return 是否规则
+         */
+        static boolean isRule(PropertyMetadata metadata) {
+            return  metadata
+                .getExpand(id)
+                .map(rule::equals)
+                .orElse(false);
+        }
+    }
+
+    /**
+     * 属性读写模式
+     */
+    interface AccessMode {
+        String id = "accessMode";
+
+        //读
+        String read = "r";
+        //写
+        String write = "w";
+        //上报
+        String report = "u";
+
+        static boolean isRead(PropertyMetadata property) {
+            return property
+                .getExpand(id)
+                .map(val -> val.toString().contains(read))
+                .orElse(true);
+        }
+
+        static boolean isWrite(PropertyMetadata property) {
+            return property
+                .getExpand(id)
+                .map(val -> val.toString().contains(write))
+                .orElseGet(() -> property
+                    .getExpand("readOnly")
+                    .map(readOnly -> !CastUtils.castBoolean(readOnly))
+                    .orElse(true)
+                );
+        }
+
+        static boolean isReport(PropertyMetadata property) {
+            return property
+                .getExpand(id)
+                .map(val -> val.toString().contains(report))
+                .orElse(true);
+        }
+    }
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/Version.java

@@ -8,6 +8,6 @@ public class Version {
 
     private final String edition = "community";
 
-    private final String version = "1.9.0-SNAPSHOT";
+    private final String version = "1.9.0";
 
 }

jetlinks-components/common-component/src/main/java/org/jetlinks/community/configuration/SmartDateDeserializer.java

@@ -9,6 +9,11 @@ import org.jetlinks.community.utils.TimeUtils;
 
 import java.util.Date;
 
+/**
+ * 时间反序列化配置
+ *
+ * @author zhouhao
+ */
 public class SmartDateDeserializer extends JsonDeserializer<Date> {
     @Override
     @SneakyThrows

jetlinks-components/common-component/src/main/java/org/jetlinks/community/micrometer/MeterRegistryManager.java

@@ -7,7 +7,6 @@ import lombok.Setter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/DateMathParser.java

@@ -21,7 +21,6 @@ package org.jetlinks.community.utils;
 
 
 import lombok.SneakyThrows;
-import org.hswebframework.utils.time.DateFormatter;
 import org.jetlinks.core.metadata.types.DateTimeType;
 import org.springframework.util.StringUtils;
 
@@ -30,6 +29,11 @@ import java.time.temporal.TemporalAdjusters;
 import java.util.function.LongSupplier;
 
 
+/**
+ * 时间转换工具
+ *
+ * @author zhouhao
+ */
 public class DateMathParser {
 
 

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/ErrorUtils.java

@@ -5,17 +5,15 @@ import org.hswebframework.web.exception.NotFoundException;
 import reactor.core.publisher.Mono;
 
 /**
+ * 异常处理工具
+ *
  * @author wangzheng
  * @see
  * @since 1.0
  */
 public class ErrorUtils {
 
-    public static <T> Mono<T> notFound(String message){
-        return Mono.error(()->new NotFoundException(message));
-    }
-
-    public static <T> Mono<T> accessDeny(String message){
-        return Mono.error(()->new AccessDenyException(message));
+    public static <T> Mono<T> notFound(String message) {
+        return Mono.error(() -> new NotFoundException(message));
     }
 }

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/TimeUtils.java

@@ -5,6 +5,11 @@ import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Date;
 
+/**
+ * 时间工具类
+ *
+ * @author zhouhao
+ */
 public class TimeUtils {
 
 

jetlinks-components/common-component/src/main/java/org/jetlinks/community/web/ErrorControllerAdvice.java

@@ -0,0 +1,106 @@
+package org.jetlinks.community.web;
+
+import com.fasterxml.jackson.databind.JsonMappingException;
+import lombok.extern.slf4j.Slf4j;
+import org.hswebframework.ezorm.rdb.exception.DuplicateKeyException;
+import org.hswebframework.ezorm.rdb.metadata.RDBColumnMetadata;
+import org.hswebframework.web.crud.web.ResponseMessage;
+import org.hswebframework.web.i18n.LocaleUtils;
+import org.jetlinks.core.enums.ErrorCode;
+import org.jetlinks.core.exception.DeviceOperationException;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.core.codec.DecodingException;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestControllerAdvice;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@RestControllerAdvice
+@Slf4j
+@Order(Ordered.HIGHEST_PRECEDENCE)
+public class ErrorControllerAdvice {
+
+    @ExceptionHandler
+    @ResponseStatus(HttpStatus.BAD_REQUEST)
+    public Mono<ResponseMessage<?>> handleException(DecodingException decodingException) {
+        Throwable cause = decodingException.getCause();
+        if (cause != null) {
+            if (cause instanceof JsonMappingException) {
+                return handleException(((JsonMappingException) cause));
+            }
+
+            return Mono.just(ResponseMessage.error(400, "illegal_argument", cause.getMessage()));
+        }
+        return Mono.just(ResponseMessage.error(400, "illegal_argument", decodingException.getMessage()));
+    }
+
+    @ExceptionHandler
+    @ResponseStatus(HttpStatus.BAD_REQUEST)
+    public Mono<ResponseMessage<?>> handleException(JsonMappingException decodingException) {
+        Throwable cause = decodingException.getCause();
+        if (cause != null) {
+            if (cause instanceof IllegalArgumentException) {
+                return Mono.just(ResponseMessage.error(400, "illegal_argument", cause.getMessage()));
+            }
+            return Mono.just(ResponseMessage.error(400, "illegal_argument", cause.getMessage()));
+        }
+        return Mono.just(ResponseMessage.error(400, "illegal_argument", decodingException.getMessage()));
+    }
+
+    @ExceptionHandler
+    @ResponseStatus(HttpStatus.BAD_REQUEST)
+    public Mono<ResponseMessage<?>> handleException(DuplicateKeyException e) {
+        List<String> columns = e
+            .getColumns()
+            .stream()
+            .map(RDBColumnMetadata::getAlias)
+            .collect(Collectors.toList());
+        if (columns.isEmpty()) {
+            return LocaleUtils
+                .resolveMessageReactive("error.duplicate_key")
+                .map(msg -> ResponseMessage.error(400, "duplicate_key", msg));
+        }
+        return LocaleUtils
+            .resolveMessageReactive("error.duplicate_key_detail", columns)
+            .map(msg -> ResponseMessage.error(400, "duplicate_key", msg).result(columns));
+    }
+
+    @ExceptionHandler
+    public Mono<ResponseEntity<ResponseMessage<Object>>> handleException(DeviceOperationException e) {
+
+        //200
+        if (e.getCode() == ErrorCode.REQUEST_HANDLING) {
+            return LocaleUtils
+                .resolveMessageReactive("message.device_message_handing")
+                .map(msg -> ResponseEntity
+                    .status(HttpStatus.OK)
+                    .body(ResponseMessage.error(200, "request_handling", msg).result(msg))
+                );
+        }
+        HttpStatus _status = HttpStatus.INTERNAL_SERVER_ERROR;
+
+        if (e.getCode() == ErrorCode.FUNCTION_UNDEFINED
+            || e.getCode() == ErrorCode.PARAMETER_UNDEFINED) {
+            //404
+            _status = HttpStatus.NOT_FOUND;
+        } else if (e.getCode() == ErrorCode.PARAMETER_UNDEFINED) {
+            //400
+            _status = HttpStatus.BAD_REQUEST;
+        }
+
+        HttpStatus status = _status;
+        return LocaleUtils
+            .resolveMessageReactive(e.getCode().getText())
+            .map(msg -> ResponseEntity
+                .status(status)
+                .body(ResponseMessage.error(status.value(), e.getCode().name().toLowerCase(), msg))
+            );
+    }
+
+}

jetlinks-components/dashboard-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

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

@@ -3,7 +3,6 @@ package org.jetlinks.community.dashboard;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
 
 /**
  * 指标维度,如: 每小时,服务器1

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/dashboard-component/src/main/java/org/jetlinks/community/dashboard/measurements/JvmCpuMeasurementProvider.java

@@ -1,13 +1,13 @@
 package org.jetlinks.community.dashboard.measurements;
 
 import org.hswebframework.utils.time.DateFormatter;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.types.DoubleType;
 import org.jetlinks.core.metadata.unit.UnifyUnit;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 

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

@@ -3,18 +3,17 @@ package org.jetlinks.community.dashboard.measurements;
 import lombok.Getter;
 import lombok.Setter;
 import org.hswebframework.utils.time.DateFormatter;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.SimplePropertyMetadata;
 import org.jetlinks.core.metadata.types.DoubleType;
 import org.jetlinks.core.metadata.types.LongType;
 import org.jetlinks.core.metadata.types.ObjectType;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 import java.lang.management.ManagementFactory;
 import java.lang.management.MemoryMXBean;

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

@@ -1,13 +1,13 @@
 package org.jetlinks.community.dashboard.measurements;
 
 import org.hswebframework.utils.time.DateFormatter;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.types.DoubleType;
 import org.jetlinks.core.metadata.unit.UnifyUnit;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 

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

@@ -3,20 +3,18 @@ package org.jetlinks.community.dashboard.measurements;
 import lombok.Getter;
 import lombok.Setter;
 import org.hswebframework.utils.time.DateFormatter;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.SimplePropertyMetadata;
 import org.jetlinks.core.metadata.types.DoubleType;
 import org.jetlinks.core.metadata.types.LongType;
 import org.jetlinks.core.metadata.types.ObjectType;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 
-import java.lang.management.ManagementFactory;
-import java.lang.management.MemoryMXBean;
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.util.Date;

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

@@ -3,7 +3,6 @@ package org.jetlinks.community.dashboard.measurements;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.SneakyThrows;
-import reactor.core.publisher.Flux;
 
 import java.lang.management.ManagementFactory;
 import java.lang.management.OperatingSystemMXBean;

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

@@ -5,7 +5,10 @@ import org.hswebframework.web.authorization.annotation.Authorize;
 import org.hswebframework.web.authorization.annotation.QueryAction;
 import org.hswebframework.web.authorization.annotation.Resource;
 import org.hswebframework.web.exception.NotFoundException;
-import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.DashboardManager;
+import org.jetlinks.community.dashboard.DashboardObject;
+import org.jetlinks.community.dashboard.MeasurementParameter;
+import org.jetlinks.community.dashboard.MeasurementValue;
 import org.jetlinks.community.dashboard.web.request.DashboardMeasurementRequest;
 import org.jetlinks.community.dashboard.web.response.DashboardInfo;
 import org.jetlinks.community.dashboard.web.response.DashboardMeasurementResponse;

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/web/response/DimensionInfo.java

@@ -2,9 +2,9 @@ package org.jetlinks.community.dashboard.web.response;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.jetlinks.community.dashboard.MeasurementDimension;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
-import org.jetlinks.community.dashboard.MeasurementDimension;
 
 @Getter
 @Setter

jetlinks-components/dashboard-component/src/main/java/org/jetlinks/community/dashboard/web/response/MeasurementInfo.java

@@ -2,8 +2,8 @@ package org.jetlinks.community.dashboard.web.response;
 
 import lombok.Getter;
 import lombok.Setter;
-import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.community.dashboard.Measurement;
+import org.jetlinks.core.metadata.DataType;
 import reactor.core.publisher.Mono;
 
 import java.util.List;

jetlinks-components/elasticsearch-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/DefaultAggregationService.java

@@ -7,7 +7,7 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
+import org.elasticsearch.search.aggregations.bucket.histogram.LongBounds;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.ezorm.core.param.TermType;
@@ -27,7 +27,6 @@ import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
 import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
 import org.jetlinks.community.timeseries.query.AggregationQueryParam;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
@@ -240,8 +239,8 @@ public class DefaultAggregationService implements AggregationService {
         return structure;
     }
 
-    protected static ExtendedBounds getExtendedBounds(AggregationQueryParam param) {
-        return new ExtendedBounds(calculateStartWithTime(param), param.getEndWithTime());
+    protected static LongBounds getExtendedBounds(AggregationQueryParam param) {
+        return new LongBounds(calculateStartWithTime(param), param.getEndWithTime());
     }
 
     private static long calculateStartWithTime(AggregationQueryParam param) {

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/aggreation/bucket/BucketAggregationsStructure.java

@@ -1,7 +1,7 @@
 package org.jetlinks.community.elastic.search.aggreation.bucket;
 
 import lombok.*;
-import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
+import org.elasticsearch.search.aggregations.bucket.histogram.LongBounds;
 import org.hswebframework.utils.StringUtils;
 import org.jetlinks.community.elastic.search.aggreation.enums.BucketType;
 import org.jetlinks.community.elastic.search.aggreation.metrics.MetricsAggregationStructure;
@@ -37,7 +37,7 @@ public class BucketAggregationsStructure {
 
     private List<Ranges> ranges;
 
-    private ExtendedBounds extendedBounds;
+    private LongBounds extendedBounds;
 
     /**
      * 时间格式

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/configuration/ElasticSearchConfiguration.java

@@ -16,7 +16,6 @@ import org.jetlinks.community.elastic.search.embedded.EmbeddedElasticSearch;
 import org.jetlinks.community.elastic.search.embedded.EmbeddedElasticSearchProperties;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexProperties;
 import org.jetlinks.community.elastic.search.service.reactive.DefaultReactiveElasticsearchClient;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/ElasticSearchIndexMetadata.java

@@ -1,7 +1,7 @@
 package org.jetlinks.community.elastic.search.index;
 
-import org.jetlinks.core.metadata.PropertyMetadata;
 import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
+import org.jetlinks.core.metadata.PropertyMetadata;
 
 import java.util.List;
 import java.util.Map;

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/index/strategies/AbstractElasticSearchIndexStrategy.java

@@ -9,10 +9,6 @@ import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
 import org.elasticsearch.cluster.metadata.MappingMetadata;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.compress.CompressedXContent;
-import org.jetlinks.core.metadata.DataType;
-import org.jetlinks.core.metadata.PropertyMetadata;
-import org.jetlinks.core.metadata.SimplePropertyMetadata;
-import org.jetlinks.core.metadata.types.*;
 import org.jetlinks.community.elastic.search.enums.ElasticDateFormat;
 import org.jetlinks.community.elastic.search.enums.ElasticPropertyType;
 import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
@@ -20,6 +16,10 @@ import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexProperties;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexStrategy;
 import org.jetlinks.community.elastic.search.service.reactive.ReactiveElasticsearchClient;
+import org.jetlinks.core.metadata.DataType;
+import org.jetlinks.core.metadata.PropertyMetadata;
+import org.jetlinks.core.metadata.SimplePropertyMetadata;
+import org.jetlinks.core.metadata.types.*;
 import reactor.core.publisher.Mono;
 
 import java.util.*;

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 zhouhao
+ */
 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/DefaultReactiveElasticsearchClient.java

@@ -60,7 +60,6 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.*;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.get.GetResult;
-import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.index.seqno.SequenceNumbers;
@@ -111,12 +110,11 @@ import java.util.StringJoiner;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
-import static org.elasticsearch.rest.BaseRestHandler.INCLUDE_TYPE_NAME_PARAMETER;
 import static org.springframework.data.elasticsearch.client.util.RequestConverters.createContentType;
 
 @Slf4j
 public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearchClient {
-    private final HostProvider hostProvider;
+    private final HostProvider<?> hostProvider;
     private final RequestCreator requestCreator;
     private Supplier<HttpHeaders> headersSupplier = () -> HttpHeaders.EMPTY;
 

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

@@ -10,8 +10,8 @@ import org.elasticsearch.search.aggregations.Aggregations;
 import org.elasticsearch.search.aggregations.BucketOrder;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
-import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
 import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
+import org.elasticsearch.search.aggregations.bucket.histogram.LongBounds;
 import org.elasticsearch.search.aggregations.bucket.terms.Terms;
 import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
@@ -293,9 +293,9 @@ public class ReactiveAggregationService implements AggregationService {
         return queryParam;
     }
 
-    protected static ExtendedBounds getExtendedBounds(AggregationQueryParam param) {
+    protected static LongBounds getExtendedBounds(AggregationQueryParam param) {
 
-        return new ExtendedBounds(calculateStartWithTime(param), param.getEndWithTime());
+        return new LongBounds(calculateStartWithTime(param), param.getEndWithTime());
     }
 
     //聚合查询默认的时间间隔

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;
@@ -57,6 +57,8 @@ import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 /**
+ * 响应式ES数据库操作类
+ *
  * @author zhouhao
  * @since 1.0
  **/
@@ -66,20 +68,23 @@ import java.util.stream.Collectors;
 @ConfigurationProperties(prefix = "elasticsearch")
 public class ReactiveElasticSearchService implements ElasticSearchService {
 
-    private final ReactiveElasticsearchClient restClient;
-
-    private final ElasticSearchIndexManager indexManager;
-
-    private FluxSink<Buffer> sink;
-
     public static final IndicesOptions indexOptions = IndicesOptions.fromOptions(
         true, true, false, false
     );
+    //使用对象池处理Buffer,减少GC消耗
+    static ObjectPool<Buffer> pool = ObjectPool.newPool(Buffer::new);
 
     static {
         DateFormatter.supportFormatter.add(new DefaultDateFormatter(Pattern.compile("[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}.+"), "yyyy-MM-dd'T'HH:mm:ss.SSSZ"));
     }
 
+    private final ReactiveElasticsearchClient restClient;
+    private final ElasticSearchIndexManager indexManager;
+    private FluxSink<Buffer> sink;
+    @Getter
+    @Setter
+    private BufferConfig buffer = new BufferConfig();
+
     public ReactiveElasticSearchService(ReactiveElasticsearchClient restClient,
                                         ElasticSearchIndexManager indexManager) {
         this.restClient = restClient;
@@ -119,12 +124,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)
@@ -134,16 +141,16 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
     @Override
     public <T> Mono<PagerResult<T>> queryPager(String[] index, QueryParam queryParam, Function<Map<String, Object>, T> mapper) {
         return this.doQuery(index, queryParam)
-                   .flatMap(tp2 -> this
-                       .convertQueryResult(tp2.getT1(), tp2.getT2(), mapper)
-                       .collectList()
-                       .filter(CollectionUtils::isNotEmpty)
-                       .map(list -> PagerResult.of((int) tp2
-                           .getT2()
-                           .getHits()
-                           .getTotalHits().value, list, queryParam))
-                   )
-                   .switchIfEmpty(Mono.fromSupplier(PagerResult::empty));
+            .flatMap(tp2 -> this
+                .convertQueryResult(tp2.getT1(), tp2.getT2(), mapper)
+                .collectList()
+                .filter(CollectionUtils::isNotEmpty)
+                .map(list -> PagerResult.of((int) tp2
+                    .getT2()
+                    .getHits()
+                    .getTotalHits().value, list, queryParam))
+            )
+            .switchIfEmpty(Mono.fromSupplier(PagerResult::empty));
     }
 
     private <T> Flux<T> convertQueryResult(List<ElasticSearchIndexMetadata> indexList,
@@ -162,8 +169,8 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
                 }
                 return mapper
                     .apply(Optional
-                               .ofNullable(metadata.get(hit.getIndex())).orElse(indexList.get(0))
-                               .convertFromElastic(hitMap));
+                        .ofNullable(metadata.get(hit.getIndex())).orElse(indexList.get(0))
+                        .convertFromElastic(hitMap));
             });
 
     }
@@ -184,7 +191,6 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
             });
     }
 
-
     @Override
     public Mono<Long> count(String[] index, QueryParam queryParam) {
         QueryParam param = queryParam.clone();
@@ -223,8 +229,8 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
     @Override
     public <T> Mono<Void> commit(String index, Publisher<T> data) {
         return Flux.from(data)
-                   .flatMap(d -> commit(index, d))
-                   .then();
+            .flatMap(d -> commit(index, d))
+            .then();
     }
 
     @Override
@@ -235,10 +241,10 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
     @Override
     public <T> Mono<Void> save(String index, Publisher<T> data) {
         return Flux.from(data)
-                   .map(v -> Buffer.of(index, v))
-                   .collectList()
-                   .flatMap(this::doSave)
-                   .then();
+            .map(v -> Buffer.of(index, v))
+            .collectList()
+            .flatMap(this::doSave)
+            .then();
     }
 
     @Override
@@ -251,32 +257,6 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
         sink.complete();
     }
 
-    @Getter
-    @Setter
-    private BufferConfig buffer = new BufferConfig();
-
-    @Getter
-    @Setter
-    public static class BufferConfig {
-        //最小间隔
-        private int rate = Integer.getInteger("elasticsearch.buffer.rate", 1000);
-        //缓冲最大数量
-        private int bufferSize = Integer.getInteger("elasticsearch.buffer.size", 3000);
-        //缓冲超时时间
-        private Duration bufferTimeout = Duration.ofSeconds(Integer.getInteger("elasticsearch.buffer.timeout", 3));
-        //背压堆积数量限制.
-        private int bufferBackpressure = Integer.getInteger("elasticsearch.buffer.backpressure", Runtime
-            .getRuntime()
-            .availableProcessors());
-        //最大缓冲字节
-        private DataSize bufferBytes = DataSize.parse(System.getProperty("elasticsearch.buffer.bytes", "15MB"));
-
-        //最大重试次数
-        private int maxRetry = 3;
-        //重试间隔
-        private Duration minBackoff = Duration.ofSeconds(3);
-    }
-
     //@PostConstruct
     public void init() {
         int flushRate = buffer.rate;
@@ -288,10 +268,10 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
 
         FluxUtils
             .bufferRate(Flux.<Buffer>create(sink -> this.sink = sink),
-                        flushRate,
-                        bufferSize,
-                        bufferTimeout,
-                        (b, l) -> bufferedBytes.addAndGet(b.numberOfBytes()) >= bufferBytes)
+                flushRate,
+                bufferSize,
+                bufferTimeout,
+                (b, l) -> bufferedBytes.addAndGet(b.numberOfBytes()) >= bufferBytes)
             .doOnNext(buf -> bufferedBytes.set(0))
             .onBackpressureBuffer(bufferBackpressure, drop -> {
                 // TODO: 2020/11/25 将丢弃的数据存储到本地磁盘
@@ -319,56 +299,10 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
             })
             .onErrorResume((err) -> Mono
                 .fromRunnable(() -> System.err.println("保存ElasticSearch数据失败:\n" +
-                                                           org.hswebframework.utils.StringUtils.throwable2String(err))))
+                    org.hswebframework.utils.StringUtils.throwable2String(err))))
             .subscribe();
     }
 
-    //使用对象池处理Buffer,减少GC消耗
-    static ObjectPool<Buffer> pool = ObjectPool.newPool(Buffer::new);
-
-    @Getter
-    static class Buffer {
-        String index;
-        String id;
-        String payload;
-        final ObjectPool.Handle<Buffer> handle;
-
-        public Buffer(ObjectPool.Handle<Buffer> handle) {
-            this.handle = handle;
-        }
-
-        public static Buffer of(String index, Object payload) {
-            Buffer buffer;
-            try {
-                buffer = pool.get();
-            } catch (Exception e) {
-                buffer = new Buffer(null);
-            }
-            buffer.index = index;
-            Map<String, Object> data = payload instanceof Map
-                ? ((Map) payload) :
-                FastBeanCopier.copy(payload, HashMap::new);
-            Object id = data.get("id");
-            buffer.id = id == null ? null : String.valueOf(id);
-            buffer.payload = JSON.toJSONString(data);
-            return buffer;
-        }
-
-        void release() {
-            this.index = null;
-            this.id = null;
-            this.payload = null;
-            if (null != handle) {
-                handle.recycle(this);
-            }
-        }
-
-        int numberOfBytes() {
-            return payload == null ? 0 : payload.length() * 2;
-        }
-    }
-
-
     private Mono<String> getIndexForSave(String index) {
         return indexManager
             .getIndexStrategy(index)
@@ -385,48 +319,48 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
 
     protected Mono<Integer> doSave(Collection<Buffer> buffers) {
         return Flux.fromIterable(buffers)
-                   .groupBy(Buffer::getIndex,Integer.MAX_VALUE)
-                   .flatMap(group -> {
-                       String index = group.key();
-                       return this
-                           .getIndexForSave(index)
-                           .flatMapMany(realIndex -> group
-                               .map(buffer -> {
-                                   try {
-                                       IndexRequest request;
-                                       if (buffer.id != null) {
-                                           request = new IndexRequest(realIndex).type("_doc").id(buffer.id);
-                                       } else {
-                                           request = new IndexRequest(realIndex).type("_doc");
-                                       }
-                                       request.source(buffer.payload, XContentType.JSON);
-                                       return request;
-                                   } finally {
-                                       buffer.release();
-                                   }
-                               }));
-                   })
-                   .collectList()
-                   .filter(CollectionUtils::isNotEmpty)
-                   .flatMap(lst -> {
-                       BulkRequest request = new BulkRequest();
-                       request.timeout(TimeValue.timeValueSeconds(9));
-                       lst.forEach(request::add);
-                       return restClient
-                           .bulk(request)
-                           .as(save -> {
-                               if (buffer.maxRetry > 0) {
-                                   return save.retryWhen(Retry.backoff(buffer.maxRetry, buffer.minBackoff));
-                               }
-                               return save;
-                           });
-                   })
-                   .doOnNext(response -> {
-                       if (response.hasFailures()) {
-                           System.err.println(response.buildFailureMessage());
-                       }
-                   })
-                   .thenReturn(buffers.size());
+            .groupBy(Buffer::getIndex, Integer.MAX_VALUE)
+            .flatMap(group -> {
+                String index = group.key();
+                return this
+                    .getIndexForSave(index)
+                    .flatMapMany(realIndex -> group
+                        .map(buffer -> {
+                            try {
+                                IndexRequest request;
+                                if (buffer.id != null) {
+                                    request = new IndexRequest(realIndex).type("_doc").id(buffer.id);
+                                } else {
+                                    request = new IndexRequest(realIndex).type("_doc");
+                                }
+                                request.source(buffer.payload, XContentType.JSON);
+                                return request;
+                            } finally {
+                                buffer.release();
+                            }
+                        }));
+            })
+            .collectList()
+            .filter(CollectionUtils::isNotEmpty)
+            .flatMap(lst -> {
+                BulkRequest request = new BulkRequest();
+                request.timeout(TimeValue.timeValueSeconds(9));
+                lst.forEach(request::add);
+                return restClient
+                    .bulk(request)
+                    .as(save -> {
+                        if (buffer.maxRetry > 0) {
+                            return save.retryWhen(Retry.backoff(buffer.maxRetry, buffer.minBackoff));
+                        }
+                        return save;
+                    });
+            })
+            .doOnNext(response -> {
+                if (response.hasFailures()) {
+                    System.err.println(response.buildFailureMessage());
+                }
+            })
+            .thenReturn(buffers.size());
     }
 
     @SneakyThrows
@@ -442,14 +376,14 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
 
     private <T> List<T> translate(Function<Map<String, Object>, T> mapper, SearchResponse response) {
         return Arrays.stream(response.getHits().getHits())
-                     .map(hit -> {
-                         Map<String, Object> hitMap = hit.getSourceAsMap();
-                         if (StringUtils.isEmpty(hitMap.get("id"))) {
-                             hitMap.put("id", hit.getId());
-                         }
-                         return mapper.apply(hitMap);
-                     })
-                     .collect(Collectors.toList());
+            .map(hit -> {
+                Map<String, Object> hitMap = hit.getSourceAsMap();
+                if (StringUtils.isEmpty(hitMap.get("id"))) {
+                    hitMap.put("id", hit.getId());
+                }
+                return mapper.apply(hitMap);
+            })
+            .collect(Collectors.toList());
     }
 
     private Flux<SearchHit> doSearch(SearchRequest request) {
@@ -482,12 +416,12 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
 
         SearchSourceBuilder builder = ElasticSearchConverter.convertSearchSourceBuilder(queryParam, indexes.get(0));
         return Flux.fromIterable(indexes)
-                   .flatMap(index -> getIndexForSearch(index.getIndex()))
-                   .collectList()
-                   .map(indexList ->
-                            new SearchRequest(indexList.toArray(new String[0]))
-                                .source(builder)
-                                .indicesOptions(indexOptions));
+            .flatMap(index -> getIndexForSearch(index.getIndex()))
+            .collectList()
+            .map(indexList ->
+                new SearchRequest(indexList.toArray(new String[0]))
+                    .source(builder)
+                    .indicesOptions(indexOptions));
     }
 
     protected Mono<QueryBuilder> createQueryBuilder(QueryParam queryParam, String index) {
@@ -504,9 +438,9 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
 
         SearchSourceBuilder builder = ElasticSearchConverter.convertSearchSourceBuilder(queryParam, indexes.get(0));
         return Flux.fromIterable(indexes)
-                   .flatMap(index -> getIndexForSearch(index.getIndex()))
-                   .collectList()
-                   .map(indexList -> new CountRequest(indexList.toArray(new String[0])).source(builder));
+            .flatMap(index -> getIndexForSearch(index.getIndex()))
+            .collectList()
+            .map(indexList -> new CountRequest(indexList.toArray(new String[0])).source(builder));
     }
 
     private Mono<CountRequest> createCountRequest(QueryParam queryParam, String... index) {
@@ -516,4 +450,68 @@ public class ReactiveElasticSearchService implements ElasticSearchService {
             .filter(CollectionUtils::isNotEmpty)
             .flatMap(list -> createCountRequest(queryParam, list));
     }
+
+    @Getter
+    @Setter
+    public static class BufferConfig {
+        //最小间隔
+        private int rate = Integer.getInteger("elasticsearch.buffer.rate", 1000);
+        //缓冲最大数量
+        private int bufferSize = Integer.getInteger("elasticsearch.buffer.size", 3000);
+        //缓冲超时时间
+        private Duration bufferTimeout = Duration.ofSeconds(Integer.getInteger("elasticsearch.buffer.timeout", 3));
+        //背压堆积数量限制.
+        private int bufferBackpressure = Integer.getInteger("elasticsearch.buffer.backpressure", Runtime
+            .getRuntime()
+            .availableProcessors());
+        //最大缓冲字节
+        private DataSize bufferBytes = DataSize.parse(System.getProperty("elasticsearch.buffer.bytes", "15MB"));
+
+        //最大重试次数
+        private int maxRetry = 3;
+        //重试间隔
+        private Duration minBackoff = Duration.ofSeconds(3);
+    }
+
+    @Getter
+    static class Buffer {
+        final ObjectPool.Handle<Buffer> handle;
+        String index;
+        String id;
+        String payload;
+
+        public Buffer(ObjectPool.Handle<Buffer> handle) {
+            this.handle = handle;
+        }
+
+        public static Buffer of(String index, Object payload) {
+            Buffer buffer;
+            try {
+                buffer = pool.get();
+            } catch (Exception e) {
+                buffer = new Buffer(null);
+            }
+            buffer.index = index;
+            Map<String, Object> data = payload instanceof Map
+                ? ((Map) payload) :
+                FastBeanCopier.copy(payload, HashMap::new);
+            Object id = data.get("id");
+            buffer.id = id == null ? null : String.valueOf(id);
+            buffer.payload = JSON.toJSONString(data);
+            return buffer;
+        }
+
+        void release() {
+            this.index = null;
+            this.id = null;
+            this.payload = null;
+            if (null != handle) {
+                handle.recycle(this);
+            }
+        }
+
+        int numberOfBytes() {
+            return payload == null ? 0 : payload.length() * 2;
+        }
+    }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/timeseries/ElasticSearchTimeSeriesManager.java

@@ -1,8 +1,6 @@
 package org.jetlinks.community.elastic.search.timeseries;
 
 import lombok.extern.slf4j.Slf4j;
-import org.jetlinks.core.metadata.SimplePropertyMetadata;
-import org.jetlinks.core.metadata.types.DateTimeType;
 import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
 import org.jetlinks.community.elastic.search.service.AggregationService;
@@ -11,6 +9,8 @@ import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.community.timeseries.TimeSeriesMetadata;
 import org.jetlinks.community.timeseries.TimeSeriesMetric;
 import org.jetlinks.community.timeseries.TimeSeriesService;
+import org.jetlinks.core.metadata.SimplePropertyMetadata;
+import org.jetlinks.core.metadata.types.DateTimeType;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/utils/ElasticSearchConverter.java

@@ -2,13 +2,13 @@ package org.jetlinks.community.elastic.search.utils;
 
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.hswebframework.ezorm.core.param.QueryParam;
+import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
 import org.jetlinks.core.metadata.Converter;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.PropertyMetadata;
 import org.jetlinks.core.metadata.types.DateTimeType;
 import org.jetlinks.core.metadata.types.GeoPoint;
 import org.jetlinks.core.metadata.types.GeoType;
-import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
 
 import java.util.Date;
 import java.util.HashMap;

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/utils/QueryParamTranslator.java

@@ -9,10 +9,10 @@ import org.elasticsearch.search.sort.SortOrder;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.ezorm.core.param.Sort;
 import org.hswebframework.ezorm.core.param.Term;
-import org.jetlinks.core.metadata.DataType;
-import org.jetlinks.core.metadata.PropertyMetadata;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.parser.DefaultLinkTypeParser;
+import org.jetlinks.core.metadata.DataType;
+import org.jetlinks.core.metadata.PropertyMetadata;
 import org.springframework.util.StringUtils;
 
 import java.util.Map;

jetlinks-components/gateway-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

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

@@ -1,8 +1,8 @@
 package org.jetlinks.community.gateway;
 
+import org.jetlinks.community.network.NetworkType;
 import org.jetlinks.core.message.Message;
 import org.jetlinks.core.message.codec.Transport;
-import org.jetlinks.community.network.NetworkType;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 

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

@@ -2,42 +2,97 @@ package org.jetlinks.community.gateway;
 
 import com.alibaba.fastjson.JSON;
 import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
 import org.jetlinks.core.event.TopicPayload;
+import org.jetlinks.core.message.CommonDeviceMessage;
+import org.jetlinks.core.message.CommonDeviceMessageReply;
 import org.jetlinks.core.message.DeviceMessage;
 import org.jetlinks.core.message.MessageType;
-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.message.property.*;
+import org.springframework.util.StringUtils;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 public class DeviceMessageUtils {
 
+    @SuppressWarnings("all")
     public static Optional<DeviceMessage> convert(TopicPayload message) {
         return Optional.of(message.decode(DeviceMessage.class));
     }
 
     public static Optional<DeviceMessage> convert(ByteBuf payload) {
+        try {
+            return MessageType.convertMessage(JSON.parseObject(payload.toString(StandardCharsets.UTF_8)));
+        } finally {
+            ReferenceCountUtil.safeRelease(payload);
+        }
+    }
 
-        return MessageType.convertMessage(JSON.parseObject(payload.toString(StandardCharsets.UTF_8)));
-
+    public static void trySetProperties(DeviceMessage message, Map<String, Object> properties) {
+        if (message instanceof ReportPropertyMessage) {
+            ((ReportPropertyMessage) message).setProperties(properties);
+        } else if (message instanceof ReadPropertyMessageReply) {
+            ((ReadPropertyMessageReply) message).setProperties(properties);
+        } else if (message instanceof WritePropertyMessageReply) {
+            ((WritePropertyMessageReply) message).setProperties(properties);
+        }
     }
 
     public static Optional<Map<String, Object>> tryGetProperties(DeviceMessage message) {
 
-        if (message instanceof ReportPropertyMessage) {
-            return Optional.ofNullable(((ReportPropertyMessage) message).getProperties());
+        if (message instanceof PropertyMessage) {
+            return Optional.ofNullable(((PropertyMessage) message).getProperties());
         }
 
-        if (message instanceof ReadPropertyMessageReply) {
-            return Optional.ofNullable(((ReadPropertyMessageReply) message).getProperties());
+        return Optional.empty();
+    }
+
+    public static Optional<Map<String, Long>> tryGetPropertySourceTimes(DeviceMessage message) {
+        if (message instanceof PropertyMessage) {
+            return Optional.ofNullable(((PropertyMessage) message).getPropertySourceTimes());
         }
-        if (message instanceof WritePropertyMessageReply) {
-            return Optional.ofNullable(((WritePropertyMessageReply) message).getProperties());
+        return Optional.empty();
+    }
+
+    public static Optional<Map<String, String>> tryGetPropertyStates(DeviceMessage message) {
+        if (message instanceof PropertyMessage) {
+            return Optional.ofNullable(((PropertyMessage) message).getPropertyStates());
         }
         return Optional.empty();
     }
 
+    public static List<Property> tryGetCompleteProperties(DeviceMessage message) {
+
+        if (message instanceof PropertyMessage) {
+            return ((PropertyMessage) message).getCompleteProperties();
+        }
+
+        return Collections.emptyList();
+    }
+
+    public static void trySetDeviceId(DeviceMessage message, String deviceId) {
+        if (message instanceof CommonDeviceMessage) {
+            ((CommonDeviceMessage) message).setDeviceId(deviceId);
+        } else if (message instanceof CommonDeviceMessageReply) {
+            ((CommonDeviceMessageReply<?>) message).setDeviceId(deviceId);
+        }
+    }
+
+    public static void trySetMessageId(DeviceMessage message, Supplier<String> messageId) {
+        if (StringUtils.hasText(message.getMessageId())) {
+            return;
+        }
+
+        if (message instanceof CommonDeviceMessage) {
+            ((CommonDeviceMessage) message).setMessageId(messageId.get());
+        } else if (message instanceof CommonDeviceMessageReply) {
+            ((CommonDeviceMessageReply<?>) message).setMessageId(messageId.get());
+        }
+    }
+
 }

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

@@ -2,7 +2,6 @@ package org.jetlinks.community.gateway;
 
 import io.netty.buffer.ByteBuf;
 import lombok.Getter;
-import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.rule.engine.executor.PayloadType;
 
 import javax.annotation.Nonnull;

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

@@ -1,6 +1,8 @@
 package org.jetlinks.community.gateway;
 
-import lombok.*;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
 
 import java.util.Collection;
 import java.util.stream.Collectors;

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

@@ -1,182 +0,0 @@
-package org.jetlinks.community.gateway;
-
-import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.Setter;
-import org.springframework.util.AntPathMatcher;
-import org.springframework.util.StringUtils;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.function.BiFunction;
-
-@Getter
-@Setter
-@EqualsAndHashCode(of = "part")
-public class TopicPart {
-
-    private TopicPart parent;
-
-    private String part;
-
-    private volatile String topic;
-
-    private int depth;
-
-    private ConcurrentMap<String, TopicPart> child = new ConcurrentHashMap<>();
-
-    private Set<String> sessionId = new CopyOnWriteArraySet<>();
-
-    private static final AntPathMatcher matcher = new AntPathMatcher();
-
-    public TopicPart(TopicPart parent, String part) {
-
-        if (StringUtils.isEmpty(part) || part.equals("/")) {
-            this.part = "";
-        } else {
-            if (part.contains("/")) {
-                this.ofTopic(part);
-            } else {
-                this.part = part;
-            }
-        }
-        this.parent = parent;
-        if (null != parent) {
-            this.depth = parent.depth + 1;
-        }
-    }
-
-    public String getTopic() {
-        if (topic == null) {
-            TopicPart parent = getParent();
-            StringBuilder builder = new StringBuilder();
-            if (parent != null) {
-                String parentTopic = parent.getTopic();
-                builder.append(parentTopic).append(parentTopic.equals("/") ? "" : "/");
-            } else {
-                builder.append("/");
-            }
-            return topic = builder.append(part).toString();
-        }
-        return topic;
-    }
-
-
-    public TopicPart subscribe(String topic) {
-        return getOrDefault(topic, TopicPart::new);
-    }
-
-    public void addSessionId(String... sessionId) {
-        this.sessionId.addAll(Arrays.asList(sessionId));
-    }
-
-    public void removeSession(String... sessionId) {
-        this.sessionId.removeAll(Arrays.asList(sessionId));
-    }
-
-    private void ofTopic(String topic) {
-        String[] parts = topic.split("[/]", 2);
-        this.part = parts[0];
-        if (parts.length > 1) {
-            TopicPart part = new TopicPart(this, parts[1]);
-            this.child.put(part.part, part);
-        }
-    }
-
-    private TopicPart getOrDefault(String topic, BiFunction<TopicPart, String, TopicPart> mapping) {
-        if (topic.startsWith("/")) {
-            topic = topic.substring(1);
-        }
-        String[] parts = topic.split("[/]");
-        TopicPart part = child.computeIfAbsent(parts[0], _topic -> mapping.apply(this, _topic));
-        for (int i = 1; i < parts.length && part != null; i++) {
-            TopicPart parent = part;
-            part = part.child.computeIfAbsent(parts[i], _topic -> mapping.apply(parent, _topic));
-        }
-        return part;
-    }
-
-    public Mono<TopicPart> get(String topic) {
-        return Mono.justOrEmpty(getOrDefault(topic, ((topicPart, s) -> null)));
-    }
-
-    public Flux<TopicPart> find(String topic) {
-        return find(topic, this);
-    }
-
-    @Override
-    public String toString() {
-        return "topic: " + getTopic() + ", sessions: " + sessionId.size();
-    }
-
-
-    public static Flux<TopicPart> find(String topic,
-                                       TopicPart topicPart) {
-        return Flux.create(sink -> {
-            ArrayDeque<TopicPart> cache = new ArrayDeque<>();
-            cache.add(topicPart);
-            String[] topicParts = topic.split("[/]");
-            String nextPart = null;
-            while (!cache.isEmpty() && !sink.isCancelled()) {
-                TopicPart part = cache.poll();
-                if (part == null) {
-                    break;
-                }
-                if (part.part.equals("**")
-//                    || part.part.equals("*")
-                    || matcher.match(part.getTopic(), topic)
-                    || (matcher.match(topic, part.getTopic()))) {
-                    sink.next(part);
-                }
-
-                //订阅了如 /device/**/event/*
-                if (part.part.equals("**")) {
-                    TopicPart tmp = null;
-                    for (int i = part.depth; i < topicParts.length; i++) {
-                        tmp = part.child.get(topicParts[i]);
-                        if (tmp != null) {
-                            cache.add(tmp);
-                        }
-                    }
-                    if (null != tmp) {
-                        continue;
-                    }
-                }
-                if ("**".equals(nextPart) || "*".equals(nextPart)) {
-                    cache.addAll(part.child.values());
-                    continue;
-                }
-                TopicPart next = part.child.get("**");
-                if (next != null) {
-                    cache.add(next);
-                }
-                next = part.child.get("*");
-                if (next != null) {
-                    cache.add(next);
-                }
-
-                if (part.depth + 1 >= topicParts.length) {
-                    continue;
-                }
-                nextPart = topicParts[part.depth + 1];
-                if (nextPart.equals("*") || nextPart.equals("**")) {
-                    cache.addAll(part.child.values());
-                    continue;
-                }
-                next = part.child.get(nextPart);
-                if (next != null) {
-                    cache.add(next);
-                }
-
-            }
-            sink.complete();
-        });
-
-    }
-
-
-}

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/external/Message.java

@@ -22,6 +22,11 @@ public interface Message {
         return new SimpleMessage(id, topic, null, Type.error, message);
     }
 
+    static Message error(String id, String topic, Throwable message) {
+
+        return new SimpleMessage(id, topic, null, Type.error, message.getMessage() == null ? message.getClass().getSimpleName() : message.getMessage());
+    }
+
     static Message success(String id, String topic, Object payload) {
         return new SimpleMessage(id, topic, payload, Type.result, null);
     }
@@ -30,11 +35,16 @@ public interface Message {
         return new SimpleMessage(id, null, null, Type.complete, null);
     }
 
+    static Message pong(String id) {
+        return new SimpleMessage(id, null, null, Type.pong, null);
+    }
+
     enum Type {
         authError,
         result,
         error,
-        complete
+        complete,
+        ping,
+        pong
     }
-
 }

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/external/socket/WebSocketMessagingHandler.java

@@ -3,25 +3,29 @@ package org.jetlinks.community.gateway.external.socket;
 import com.alibaba.fastjson.JSON;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.hswebframework.web.authorization.Authentication;
 import org.hswebframework.web.authorization.ReactiveAuthenticationManager;
 import org.hswebframework.web.authorization.token.UserToken;
 import org.hswebframework.web.authorization.token.UserTokenManager;
+import org.hswebframework.web.logger.ReactiveLogger;
 import org.jetlinks.community.gateway.external.Message;
 import org.jetlinks.community.gateway.external.MessagingManager;
 import org.jetlinks.community.gateway.external.SubscribeRequest;
+import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.util.StringUtils;
 import org.springframework.web.reactive.socket.CloseStatus;
 import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.WebSocketMessage;
 import org.springframework.web.reactive.socket.WebSocketSession;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
+import reactor.util.context.Context;
 
 import javax.annotation.Nonnull;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentSkipListSet;
 
 @AllArgsConstructor
 @Slf4j
@@ -60,8 +64,25 @@ public class WebSocketMessagingHandler implements WebSocketHandler {
                 .receive()
                 .doOnNext(message -> {
                     try {
+                        if (message.getType() == WebSocketMessage.Type.PONG) {
+                            return;
+                        }
+                        if (message.getType() == WebSocketMessage.Type.PING) {
+                            session
+                                .send(Mono.just(session.pongMessage(DataBufferFactory::allocateBuffer)))
+                                .subscribe();
+                            return;
+                        }
                         MessagingRequest request = JSON.parseObject(message.getPayloadAsText(), MessagingRequest.class);
-                        if (request == null || request.getType() == MessagingRequest.Type.ping) {
+                        if (request == null) {
+                            return;
+                        }
+                        if (request.getType() == MessagingRequest.Type.ping) {
+                            session
+                                .send(Mono.just(session.textMessage(JSON.toJSONString(
+                                    Message.pong(request.getId())
+                                ))))
+                                .subscribe();
                             return;
                         }
                         if (StringUtils.isEmpty(request.getId())) {
@@ -73,33 +94,46 @@ public class WebSocketMessagingHandler implements WebSocketHandler {
                         }
                         if (request.getType() == MessagingRequest.Type.sub) {
                             //重复订阅
-                            if (subs.containsKey(request.getId())) {
+                            Disposable old = subs.get(request.getId());
+                            if (old != null && !old.isDisposed()) {
                                 return;
                             }
-                            subs.put(request.getId(), messagingManager
+                            Map<String, String> context = new HashMap<>();
+                            context.put("userId", auth.getUser().getId());
+                            context.put("userName", auth.getUser().getName());
+                            Disposable sub = messagingManager
                                 .subscribe(SubscribeRequest.of(request, auth))
-                                .onErrorResume(err -> Mono.just(Message.error(request.getId(), request.getTopic(), err.getMessage())))
+                                .doOnEach(ReactiveLogger.onError(err -> log.error("{}", err.getMessage(), err)))
+                                .onErrorResume(err -> Mono.just(Message.error(request.getId(), request.getTopic(), err)))
                                 .map(msg -> session.textMessage(JSON.toJSONString(msg)))
                                 .doOnComplete(() -> {
+                                    log.debug("complete subscription:{}", request.getTopic());
                                     subs.remove(request.getId());
                                     Mono.just(session.textMessage(JSON.toJSONString(Message.complete(request.getId()))))
                                         .as(session::send)
                                         .subscribe();
                                 })
-                                .flatMap(msg -> session.send(Mono.just(msg)))
-                                .subscribe()
-                            );
-
+                                .doOnCancel(() -> {
+                                    log.debug("cancel subscription:{}", request.getTopic());
+                                    subs.remove(request.getId());
+                                })
+                                .transform(session::send)
+                                .subscriberContext(ReactiveLogger.start(context))
+                                .subscriberContext(Context.of(Authentication.class, auth))
+                                .subscribe();
+                            if (!sub.isDisposed()) {
+                                subs.put(request.getId(), sub);
+                            }
                         } else if (request.getType() == MessagingRequest.Type.unsub) {
                             Optional.ofNullable(subs.remove(request.getId()))
-                                .ifPresent(Disposable::dispose);
+                                    .ifPresent(Disposable::dispose);
                         } else {
                             session.send(Mono.just(session.textMessage(JSON.toJSONString(
                                 Message.error(request.getId(), request.getTopic(), "不支持的类型:" + request.getType())
                             )))).subscribe();
                         }
                     } catch (Exception e) {
-                        log.warn(e.getMessage(),e);
+                        log.warn(e.getMessage(), e);
                         session.send(Mono.just(session.textMessage(JSON.toJSONString(
                             Message.error("illegal_argument", null, "消息格式错误")
                         )))).subscribe();

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerDeviceGatewayMonitor.java

@@ -3,7 +3,6 @@ package org.jetlinks.community.gateway.monitor;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
-import lombok.AllArgsConstructor;
 
 import java.util.concurrent.atomic.AtomicReference;
 

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurement.java

@@ -2,7 +2,12 @@ package org.jetlinks.community.gateway.monitor.measurements;
 
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
 import org.jetlinks.community.Interval;
+import org.jetlinks.community.dashboard.*;
+import org.jetlinks.community.dashboard.supports.StaticMeasurement;
+import org.jetlinks.community.gateway.monitor.GatewayTimeSeriesMetric;
+import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.community.timeseries.query.Aggregation;
+import org.jetlinks.community.timeseries.query.AggregationQueryParam;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.DefaultConfigMetadata;
@@ -10,14 +15,8 @@ import org.jetlinks.core.metadata.types.DateTimeType;
 import org.jetlinks.core.metadata.types.EnumType;
 import org.jetlinks.core.metadata.types.IntType;
 import org.jetlinks.core.metadata.types.StringType;
-import org.jetlinks.community.dashboard.*;
-import org.jetlinks.community.dashboard.supports.StaticMeasurement;
-import org.jetlinks.community.gateway.monitor.GatewayTimeSeriesMetric;
-import org.jetlinks.community.timeseries.TimeSeriesManager;
-import org.jetlinks.community.timeseries.query.AggregationQueryParam;
 import reactor.core.publisher.Flux;
 
-import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.Date;

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurementProvider.java

@@ -1,12 +1,11 @@
 package org.jetlinks.community.gateway.monitor.measurements;
 
-import org.jetlinks.community.dashboard.MeasurementDefinition;
 import org.jetlinks.community.dashboard.supports.StaticMeasurementProvider;
 import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.community.timeseries.query.Aggregation;
 import org.springframework.stereotype.Component;
 
-import static org.jetlinks.community.dashboard.MeasurementDefinition.*;
+import static org.jetlinks.community.dashboard.MeasurementDefinition.of;
 
 @Component
 public class DeviceGatewayMeasurementProvider extends StaticMeasurementProvider {

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 zhouhao
+ */
 @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,9 +7,17 @@ import org.jetlinks.community.ValueObject;
 import java.util.HashMap;
 import java.util.Map;
 
+/**
+ * 设备网关属性外观类
+ * <p>
+ * 转换设备网关属性数据
+ * </p>
+ *
+ * @author zhouhao
+ */
 @Getter
 @Setter
-public class DeviceGatewayProperties  implements ValueObject {
+public class DeviceGatewayProperties implements ValueObject {
 
     private String id;
 
@@ -17,7 +25,7 @@ public class DeviceGatewayProperties  implements ValueObject {
 
     private String networkId;
 
-    private Map<String,Object> configuration=new HashMap<>();
+    private Map<String, Object> configuration = new HashMap<>();
 
     @Override
     public Map<String, Object> values() {

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

@@ -2,8 +2,19 @@ package org.jetlinks.community.gateway.supports;
 
 import reactor.core.publisher.Mono;
 
+/**
+ * 设备网关属性管理器
+ *
+ * @author zhouhao
+ */
 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,14 @@ import org.jetlinks.community.gateway.DeviceGateway;
 import org.jetlinks.community.network.NetworkType;
 import reactor.core.publisher.Mono;
 
+/**
+ * 设备网关支持提供商,用于提供对各种设备网关的支持.在启动设备网关时,会根据对应的提供商以及配置来创建设备网关.
+ * 实现统一管理网关配置,动态创建设备网关.
+ *
+ * @author zhouhao
+ * @see DeviceGateway
+ * @since 1.0
+ */
 public interface DeviceGatewayProvider {
 
     String getId();

jetlinks-components/io-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>
@@ -17,7 +17,7 @@
         <dependency>
             <groupId>commons-io</groupId>
             <artifactId>commons-io</artifactId>
-            <version>2.6</version>
+            <version>2.7</version>
         </dependency>
 
         <dependency>

jetlinks-components/logging-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

jetlinks-components/logging-component/src/main/java/org/jetlinks/community/logging/configuration/LoggingProperties.java

@@ -5,7 +5,10 @@ import lombok.Setter;
 import org.jetlinks.community.logging.system.SerializableSystemLog;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 @ConfigurationProperties(prefix = "jetlinks.logging")
 @Getter

jetlinks-components/logging-component/src/main/java/org/jetlinks/community/logging/event/handler/AccessLoggerEventHandler.java

@@ -1,13 +1,13 @@
 package org.jetlinks.community.logging.event.handler;
 
 import lombok.extern.slf4j.Slf4j;
-import org.jetlinks.core.metadata.types.DateTimeType;
-import org.jetlinks.core.metadata.types.ObjectType;
-import org.jetlinks.core.metadata.types.StringType;
 import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
 import org.jetlinks.community.elastic.search.service.ElasticSearchService;
 import org.jetlinks.community.logging.access.SerializableAccessLog;
+import org.jetlinks.core.metadata.types.DateTimeType;
+import org.jetlinks.core.metadata.types.ObjectType;
+import org.jetlinks.core.metadata.types.StringType;
 import org.springframework.context.event.EventListener;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;

jetlinks-components/logging-component/src/main/java/org/jetlinks/community/logging/event/handler/SystemLoggerEventHandler.java

@@ -1,14 +1,14 @@
 package org.jetlinks.community.logging.event.handler;
 
 import lombok.extern.slf4j.Slf4j;
-import org.jetlinks.core.event.EventBus;
-import org.jetlinks.core.metadata.types.DateTimeType;
-import org.jetlinks.core.metadata.types.ObjectType;
-import org.jetlinks.core.metadata.types.StringType;
 import org.jetlinks.community.elastic.search.index.DefaultElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
 import org.jetlinks.community.elastic.search.service.ElasticSearchService;
 import org.jetlinks.community.logging.system.SerializableSystemLog;
+import org.jetlinks.core.event.EventBus;
+import org.jetlinks.core.metadata.types.DateTimeType;
+import org.jetlinks.core.metadata.types.ObjectType;
+import org.jetlinks.core.metadata.types.StringType;
 import org.springframework.context.event.EventListener;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;

jetlinks-components/network-component/mqtt-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>network-component</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/client/MqttClient.java

@@ -1,7 +1,7 @@
 package org.jetlinks.community.network.mqtt.client;
 
-import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.community.network.Network;
+import org.jetlinks.core.message.codec.MqttMessage;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/client/MqttClientProvider.java

@@ -7,14 +7,14 @@ import io.vertx.mqtt.MqttClientOptions;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.utils.ExpressionUtils;
+import org.jetlinks.community.network.*;
+import org.jetlinks.community.network.security.CertificateManager;
+import org.jetlinks.community.network.security.VertxKeyCertTrustOptions;
 import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.core.metadata.DefaultConfigMetadata;
 import org.jetlinks.core.metadata.types.BooleanType;
 import org.jetlinks.core.metadata.types.IntType;
 import org.jetlinks.core.metadata.types.StringType;
-import org.jetlinks.community.network.*;
-import org.jetlinks.community.network.security.CertificateManager;
-import org.jetlinks.community.network.security.VertxKeyCertTrustOptions;
 import org.springframework.core.env.Environment;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/client/VertxMqttClient.java

@@ -5,11 +5,11 @@ import io.vertx.core.buffer.Buffer;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.MapUtils;
+import org.jetlinks.community.network.DefaultNetworkType;
+import org.jetlinks.community.network.NetworkType;
 import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.core.message.codec.SimpleMqttMessage;
 import org.jetlinks.core.topic.Topic;
-import org.jetlinks.community.network.DefaultNetworkType;
-import org.jetlinks.community.network.NetworkType;
 import reactor.core.Disposable;
 import reactor.core.Disposables;
 import reactor.core.publisher.Flux;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/executor/MqttClientTaskExecutorProvider.java

@@ -3,11 +3,11 @@ package org.jetlinks.community.network.mqtt.executor;
 import lombok.AllArgsConstructor;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.dict.EnumDict;
-import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.community.network.DefaultNetworkType;
 import org.jetlinks.community.network.NetworkManager;
 import org.jetlinks.community.network.PubSubType;
 import org.jetlinks.community.network.mqtt.client.MqttClient;
+import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.rule.engine.api.RuleConstants;
 import org.jetlinks.rule.engine.api.RuleData;
 import org.jetlinks.rule.engine.api.RuleDataCodecs;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java

@@ -2,6 +2,15 @@ package org.jetlinks.community.network.mqtt.gateway.device;
 
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.community.gateway.DeviceGateway;
+import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
+import org.jetlinks.community.gateway.monitor.GatewayMonitors;
+import org.jetlinks.community.network.DefaultNetworkType;
+import org.jetlinks.community.network.NetworkType;
+import org.jetlinks.community.network.mqtt.client.MqttClient;
+import org.jetlinks.community.network.mqtt.gateway.device.session.MqttClientSession;
+import org.jetlinks.community.network.mqtt.gateway.device.session.UnknownDeviceMqttClientSession;
+import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.device.DeviceOperator;
@@ -13,15 +22,6 @@ import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.FromDeviceMessageContext;
 import org.jetlinks.core.message.codec.Transport;
 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.network.DefaultNetworkType;
-import org.jetlinks.community.network.NetworkType;
-import org.jetlinks.community.network.mqtt.client.MqttClient;
-import org.jetlinks.community.network.mqtt.gateway.device.session.MqttClientSession;
-import org.jetlinks.community.network.mqtt.gateway.device.session.UnknownDeviceMqttClientSession;
-import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import reactor.core.Disposable;
 import reactor.core.publisher.EmitterProcessor;
@@ -51,6 +51,8 @@ public class MqttClientDeviceGateway implements DeviceGateway {
 
     private final String protocol;
 
+    private final int qos;
+
     private final ProtocolSupports protocolSupport;
 
     private final EmitterProcessor<Message> processor = EmitterProcessor.create(false);
@@ -72,7 +74,8 @@ public class MqttClientDeviceGateway implements DeviceGateway {
                                    String protocol,
                                    DeviceSessionManager sessionManager,
                                    DecodedClientMessageHandler clientMessageHandler,
-                                   List<String> topics) {
+                                   List<String> topics,
+                                   int qos) {
         this.gatewayMonitor = GatewayMonitors.getDeviceGatewayMonitor(id);
 
         this.id = Objects.requireNonNull(id, "id");
@@ -82,6 +85,7 @@ public class MqttClientDeviceGateway implements DeviceGateway {
         this.protocol = Objects.requireNonNull(protocol, "protocol");
         this.topics = Objects.requireNonNull(topics, "topics");
         this.helper = new DeviceGatewayHelper(registry, sessionManager, clientMessageHandler);
+        this.qos = qos;
     }
 
 
@@ -95,7 +99,7 @@ public class MqttClientDeviceGateway implements DeviceGateway {
         }
         disposable
             .add(mqttClient
-                     .subscribe(topics)
+                     .subscribe(topics,qos)
                      .filter((msg) -> started.get())
                      .flatMap(mqttMessage -> {
                          AtomicReference<Duration> timeoutRef = new AtomicReference<>();

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGatewayProvider.java

@@ -1,8 +1,5 @@
 package org.jetlinks.community.network.mqtt.gateway.device;
 
-import org.jetlinks.core.ProtocolSupports;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.community.gateway.DeviceGateway;
 import org.jetlinks.community.gateway.supports.DeviceGatewayProperties;
 import org.jetlinks.community.gateway.supports.DeviceGatewayProvider;
@@ -10,6 +7,9 @@ import org.jetlinks.community.network.DefaultNetworkType;
 import org.jetlinks.community.network.NetworkManager;
 import org.jetlinks.community.network.NetworkType;
 import org.jetlinks.community.network.mqtt.client.MqttClient;
+import org.jetlinks.core.ProtocolSupports;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;
@@ -64,6 +64,7 @@ public class MqttClientDeviceGatewayProvider implements DeviceGatewayProvider {
 
                 String protocol = (String) properties.getConfiguration().get("protocol");
                 String topics = (String) properties.getConfiguration().get("topics");
+                int qos = properties.getInt("qos").orElse(0);
                 Objects.requireNonNull(topics, "topics");
 
                 return new MqttClientDeviceGateway(properties.getId(),
@@ -73,7 +74,8 @@ public class MqttClientDeviceGatewayProvider implements DeviceGatewayProvider {
                                                    protocol,
                                                    sessionManager,
                                                    clientMessageHandler,
-                                                   Arrays.asList(topics.split("[,;\n]"))
+                                                   Arrays.asList(topics.split("[,;\n]")),
+                                                   qos
                 );
 
             });

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java

@@ -4,6 +4,16 @@ import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
 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.mqtt.gateway.device.session.MqttConnectionSession;
+import org.jetlinks.community.network.mqtt.server.MqttConnection;
+import org.jetlinks.community.network.mqtt.server.MqttServer;
+import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.device.AuthenticationResponse;
 import org.jetlinks.core.device.DeviceOperator;
@@ -20,16 +30,6 @@ import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.session.DeviceSession;
 import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.core.server.session.ReplaceableDeviceSession;
-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.mqtt.gateway.device.session.MqttConnectionSession;
-import org.jetlinks.community.network.mqtt.server.MqttConnection;
-import org.jetlinks.community.network.mqtt.server.MqttServer;
-import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import org.springframework.util.StringUtils;
 import reactor.core.Disposable;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGatewayProvider.java

@@ -1,8 +1,5 @@
 package org.jetlinks.community.network.mqtt.gateway.device;
 
-import org.jetlinks.core.ProtocolSupports;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.community.gateway.DeviceGateway;
 import org.jetlinks.community.gateway.supports.DeviceGatewayProperties;
 import org.jetlinks.community.gateway.supports.DeviceGatewayProvider;
@@ -10,6 +7,9 @@ import org.jetlinks.community.network.DefaultNetworkType;
 import org.jetlinks.community.network.NetworkManager;
 import org.jetlinks.community.network.NetworkType;
 import org.jetlinks.community.network.mqtt.server.MqttServer;
+import org.jetlinks.core.ProtocolSupports;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttClientSession.java

@@ -2,14 +2,14 @@ package org.jetlinks.community.network.mqtt.gateway.device.session;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
+import org.jetlinks.community.network.mqtt.client.MqttClient;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.message.codec.DefaultTransport;
 import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.session.DeviceSession;
-import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
-import org.jetlinks.community.network.mqtt.client.MqttClient;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSession.java

@@ -2,12 +2,12 @@ package org.jetlinks.community.network.mqtt.gateway.device.session;
 
 import lombok.Getter;
 import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
+import org.jetlinks.community.network.mqtt.server.MqttConnection;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.session.DeviceSession;
-import org.jetlinks.community.network.mqtt.server.MqttConnection;
 import org.jetlinks.core.server.session.ReplaceableDeviceSession;
 import reactor.core.publisher.Mono;
 

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/UnknownDeviceMqttClientSession.java

@@ -1,13 +1,13 @@
 package org.jetlinks.community.network.mqtt.gateway.device.session;
 
 import lombok.Getter;
+import org.jetlinks.community.network.mqtt.client.MqttClient;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.message.codec.DefaultTransport;
 import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.MqttMessage;
 import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.session.DeviceSession;
-import org.jetlinks.community.network.mqtt.client.MqttClient;
 import reactor.core.publisher.Mono;
 
 public class UnknownDeviceMqttClientSession implements DeviceSession {

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/server/vertx/VertxMqttConnection.java

@@ -28,7 +28,6 @@ import reactor.core.publisher.Mono;
 
 import javax.annotation.Nonnull;
 import java.net.InetSocketAddress;
-import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Optional;
 import java.util.function.Consumer;

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/server/vertx/VertxMqttServerProvider.java

@@ -7,9 +7,9 @@ import io.vertx.mqtt.MqttServerOptions;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.jetlinks.community.network.*;
-import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.community.network.security.CertificateManager;
 import org.jetlinks.community.network.security.VertxKeyCertTrustOptions;
+import org.jetlinks.core.metadata.ConfigMetadata;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;
 

jetlinks-components/network-component/network-core/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>network-component</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -19,6 +19,11 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+/**
+ * 默认网络管理器
+ *
+ * @author zhouhao
+ */
 @Component
 @Slf4j
 public class DefaultNetworkManager implements NetworkManager, BeanPostProcessor {
@@ -26,9 +31,9 @@ public class DefaultNetworkManager implements NetworkManager, BeanPostProcessor
     private final NetworkConfigManager configManager;
 
 
-    private Map<String, Map<String, Network>> store = new ConcurrentHashMap<>();
+    private final Map<String, Map<String, Network>> store = new ConcurrentHashMap<>();
 
-    private Map<String, NetworkProvider<Object>> providerSupport = new ConcurrentHashMap<>();
+    private final Map<String, NetworkProvider<Object>> providerSupport = new ConcurrentHashMap<>();
 
     public DefaultNetworkManager(NetworkConfigManager configManager) {
         this.configManager = configManager;
@@ -49,7 +54,12 @@ public class DefaultNetworkManager implements NetworkManager, BeanPostProcessor
             .subscribe(t -> this.checkNetwork());
     }
 
+    /**
+     * 检查网络 把需要加载的网络组件启动起来
+     */
     protected void checkNetwork() {
+        // 获取并过滤所有停止的网络组件
+        // 重新加载启动状态的网络组件
         Flux.fromIterable(store.values())
             .flatMapIterable(Map::values)
             .filter(i -> !i.isAlive())
@@ -84,6 +94,14 @@ public class DefaultNetworkManager implements NetworkManager, BeanPostProcessor
             .map(n -> (T) n);
     }
 
+    /**
+     * 如果store中不存在网络组件就创建，存在就重新加载
+     *
+     * @param provider   网络组件支持提供商
+     * @param id         网络组件唯一标识
+     * @param properties 网络组件配置
+     * @return 网络组件
+     */
     public Network doCreate(NetworkProvider<Object> provider, String id, Object properties) {
         return getNetworkStore(provider.getType()).compute(id, (s, network) -> {
             if (network == null) {

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

@@ -2,6 +2,11 @@ package org.jetlinks.community.network;
 
 import reactor.core.publisher.Mono;
 
+/**
+ * 网络组件配置管理器
+ *
+ * @author zhouhao
+ */
 public interface NetworkConfigManager {
 
     Mono<NetworkProperties> getConfig(NetworkType networkType, String id);

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

@@ -4,13 +4,48 @@ import reactor.core.publisher.Mono;
 
 import java.util.List;
 
+/**
+ * 网络服务管理器
+ * <p>
+ * 管理所有的网络组件
+ *
+ * @author zhouhao
+ * @since 1.0
+ */
 public interface NetworkManager {
 
+    /**
+     * 根据ID获取网络组件，否则根据type和id创建网络组件并返回
+     *
+     * @param type 网络类型
+     * @param id   网络组件id
+     * @param <T>  NetWork子类泛型
+     * @return 网络组件
+     */
     <T extends Network> Mono<T> getNetwork(NetworkType type, String id);
 
+    /**
+     * 获取所有的网络组件支持提供商
+     *
+     * @return 网络组件支持提供商
+     */
     List<NetworkProvider<?>> getProviders();
 
-   Mono<Void> reload(NetworkType type, String id);
+    /**
+     * 重新加载网络组件
+     *
+     * @param type 网络类型
+     * @param id   网络组件ID
+     * @return void
+     */
+    Mono<Void> reload(NetworkType type, String id);
 
-   Mono<Void> shutdown(NetworkType type, String id);
+    /**
+     * 停止网络组件
+     *
+     * @param type 网络类型
+     * @param id   网络组件ID
+     * @return void
+     */
+    Mono<Void> shutdown(NetworkType type, String id);
 }

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/network/security/Certificate.java

@@ -6,6 +6,11 @@ import javax.net.ssl.TrustManagerFactory;
 import javax.net.ssl.X509KeyManager;
 import java.security.cert.X509Certificate;
 
+/**
+ * 证书接口
+ *
+ * @author zhouhao
+ */
 public interface Certificate {
 
     String getId();

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/network/security/CertificateManager.java

@@ -2,6 +2,11 @@ package org.jetlinks.community.network.security;
 
 import reactor.core.publisher.Mono;
 
+/**
+ * 证书管理接口
+ *
+ * @author zhouhao
+ */
 public interface CertificateManager {
 
     Mono<Certificate> getCertificate(String id);

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/network/security/DefaultCertificate.java

@@ -13,13 +13,18 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Stream;
 
+/**
+ * 默认证书
+ *
+ * @author zhouhao
+ */
 public class DefaultCertificate implements Certificate {
 
     @Getter
-    private String id;
+    private final String id;
 
     @Getter
-    private String name;
+    private final String name;
 
     private KeyStoreHelper keyHelper;
 
@@ -131,12 +136,12 @@ public class DefaultCertificate implements Certificate {
             return EMPTY;
         }
         return Arrays.stream(trustHelper
-                .getTrustMgrs())
-                .filter(X509TrustManager.class::isInstance)
-                .map(X509TrustManager.class::cast)
-                .map(X509TrustManager::getAcceptedIssuers)
-                .flatMap(Stream::of)
-                .toArray(X509Certificate[]::new);
+            .getTrustMgrs())
+            .filter(X509TrustManager.class::isInstance)
+            .map(X509TrustManager.class::cast)
+            .map(X509TrustManager::getAcceptedIssuers)
+            .flatMap(Stream::of)
+            .toArray(X509Certificate[]::new);
     }
 
     @Override

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

@@ -2,14 +2,18 @@ package org.jetlinks.community.network.utils;
 
 import lombok.AllArgsConstructor;
 import org.jetlinks.community.PropertyConstants;
+import org.jetlinks.core.device.DeviceConfigKey;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.*;
+import org.jetlinks.core.message.state.DeviceStateCheckMessage;
+import org.jetlinks.core.message.state.DeviceStateCheckMessageReply;
 import org.jetlinks.core.server.session.ChildrenDeviceSession;
 import org.jetlinks.core.server.session.DeviceSession;
 import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.core.server.session.KeepOnlineSession;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
+import org.springframework.util.StringUtils;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
@@ -17,6 +21,14 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+/**
+ * 设备网关处理工具
+ * <p>
+ * 封装常用的设备消息处理操作
+ * </p>
+ *
+ * @author zhouhao
+ */
 @AllArgsConstructor
 public class DeviceGatewayHelper {
 
@@ -45,42 +57,84 @@ public class DeviceGatewayHelper {
     }
 
     protected Mono<Void> handleChildrenDeviceMessage(String deviceId, DeviceMessage children) {
+        if (deviceId == null
+            || children instanceof DeviceStateCheckMessage
+            || children instanceof DeviceStateCheckMessageReply
+            || children instanceof DisconnectDeviceMessage
+            || children instanceof DisconnectDeviceMessageReply) {
+            return Mono.empty();
+        }
+        if (children instanceof DeviceMessageReply) {
+            DeviceMessageReply reply = ((DeviceMessageReply) children);
+            if (!reply.isSuccess()) {
+                return Mono.empty();
+            }
+        }
         ChildrenDeviceSession deviceSession = sessionManager.getSession(deviceId, children.getDeviceId());
-        //子设备离线
-        if(children instanceof DeviceOfflineMessage){
+        //子设备离线或者注销
+        if (children instanceof DeviceOfflineMessage || children instanceof DeviceUnRegisterMessage) {
             //注销会话,这里子设备可能会收到多次离线消息
             //注销会话一次离线,消息网关转发子设备消息一次
+            if (deviceSession != null && children instanceof DeviceOfflineMessage) {
+                //忽略离线消息,因为注销会话时,会自动发送一个离线消息
+                children.addHeader(Headers.ignore, true);
+            }
             return sessionManager
-                .unRegisterChildren(deviceId,children.getDeviceId())
+                .unRegisterChildren(deviceId, children.getDeviceId())
                 .then();
         }
         if (deviceSession == null && null != children.getDeviceId()) {
-            Mono<Void> then = sessionManager
+            //忽略上线消息,因为注册会话时,会自动发送一个上线消息
+            if (children instanceof DeviceOnlineMessage) {
+                children.addHeader(Headers.ignore, true);
+            }
+            Mono<Void> registerSession = sessionManager
                 .registerChildren(deviceId, children.getDeviceId())
                 .then();
             //子设备注册
             if (isDoRegister(children)) {
-                then = Mono.delay(Duration.ofSeconds(2))
-                           .then(then);
+                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))
+                    );
             }
-            return then;
+            return registerSession;
         }
         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,
                                                     Supplier<Mono<DeviceOperator>> deviceNotFoundListener) {
         String deviceId = message.getDeviceId();
+        if (StringUtils.isEmpty(deviceId)) {
+            return Mono.empty();
+        }
         Mono<Void> then = Mono.empty();
         boolean doHandle = true;
         if (message instanceof ChildDeviceMessage) {
             DeviceMessage childrenMessage = (DeviceMessage) ((ChildDeviceMessage) message).getChildDeviceMessage();
-            then = handleChildrenDeviceMessage(deviceId,childrenMessage);
+            then = handleChildrenDeviceMessage(deviceId, childrenMessage);
         } else if (message instanceof ChildDeviceMessageReply) {
             DeviceMessage childrenMessage = (DeviceMessage) ((ChildDeviceMessageReply) message).getChildDeviceMessage();
-            then = handleChildrenDeviceMessage(deviceId,childrenMessage);
+            then = handleChildrenDeviceMessage(deviceId, childrenMessage);
         } else if (message instanceof DeviceOfflineMessage) {
             //设备离线消息
             DeviceSession session = sessionManager.unregister(deviceId);
@@ -118,19 +172,21 @@ public class DeviceGatewayHelper {
                 }))
                 .flatMap(device -> {
                     DeviceSession newSession = sessionBuilder.apply(device);
-                    //保持会话，在低功率设备上,可能无法保持mqtt长连接.
-                    if (message.getHeader(Headers.keepOnline).orElse(false)) {
-                        int timeout = message.getHeaderOrDefault(Headers.keepOnlineTimeoutSeconds);
-                        newSession = new KeepOnlineSession(newSession, Duration.ofSeconds(timeout));
-                    }
-                    sessionManager.register(newSession);
-                    sessionConsumer.accept(newSession);
-                    newSession.keepAlive();
-                    if (!(message instanceof DeviceRegisterMessage) &&
-                        !(message instanceof DeviceOnlineMessage)) {
-                        return messageHandler
-                            .handleMessage(device, message)
-                            .thenReturn(device);
+                    if (null != newSession) {
+                        //保持会话，在低功率设备上,可能无法保持mqtt长连接.
+                        if (message.getHeader(Headers.keepOnline).orElse(false)) {
+                            int timeout = message.getHeaderOrDefault(Headers.keepOnlineTimeoutSeconds);
+                            newSession = new KeepOnlineSession(newSession, Duration.ofSeconds(timeout));
+                        }
+                        sessionManager.register(newSession);
+                        sessionConsumer.accept(newSession);
+                        newSession.keepAlive();
+                        if (!(message instanceof DeviceRegisterMessage) &&
+                            !(message instanceof DeviceOnlineMessage)) {
+                            return messageHandler
+                                .handleMessage(device, message)
+                                .thenReturn(device);
+                        }
                     }
                     return Mono.just(device);
                 })
@@ -138,6 +194,16 @@ public class DeviceGatewayHelper {
                 .flatMap(then::thenReturn)
                 ;
         } else {
+            //消息中指定保存在线
+            if (message.getHeader(Headers.keepOnline).orElse(false)
+                && !(session instanceof KeepOnlineSession)) {
+                Duration timeout = message
+                    .getHeader(Headers.keepOnlineTimeoutSeconds)
+                    .map(Duration::ofSeconds)
+                    .orElse(Duration.ofHours(1));
+                //替换session
+                session = sessionManager.replace(session, new KeepOnlineSession(session, timeout));
+            }
             sessionConsumer.accept(session);
             session.keepAlive();
             if (doHandle) {

jetlinks-components/network-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jetlinks-components</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <packaging>pom</packaging>

jetlinks-components/network-component/tcp-component/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>network-component</artifactId>
         <groupId>org.jetlinks.community</groupId>
-        <version>1.9.0-SNAPSHOT</version>
+        <version>1.10.0</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/TcpClient.java

@@ -2,6 +2,7 @@ package org.jetlinks.community.network.tcp.client;
 
 import org.jetlinks.community.network.Network;
 import org.jetlinks.community.network.tcp.TcpMessage;
+import org.jetlinks.core.server.ClientConnection;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -14,7 +15,7 @@ import java.time.Duration;
  * @author zhouhao
  * @version 1.0
  */
-public interface TcpClient extends Network {
+public interface TcpClient extends Network, ClientConnection {
 
     /**
      * 获取客户端远程地址

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/TcpClientProperties.java

@@ -7,7 +7,6 @@ import org.jetlinks.community.network.tcp.parser.PayloadParserType;
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Optional;
 
 @Getter
 @Setter

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClient.java

@@ -12,6 +12,7 @@ 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.parser.PayloadParser;
+import org.jetlinks.core.message.codec.EncodedMessage;
 import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
@@ -28,27 +29,23 @@ import java.util.function.Function;
 @Slf4j
 public class VertxTcpClient implements TcpClient {
 
+    @Getter
+    private final String id;
+    private final List<Runnable> disconnectListener = new CopyOnWriteArrayList<>();
+    private final EmitterProcessor<TcpMessage> processor = EmitterProcessor.create(false);
+    private final FluxSink<TcpMessage> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
+    private final boolean serverClient;
     public volatile NetClient client;
-
     public NetSocket socket;
-
     volatile PayloadParser payloadParser;
-
-    @Getter
-    private final String id;
-
     @Setter
     private long keepAliveTimeoutMs = Duration.ofMinutes(10).toMillis();
-
     private volatile long lastKeepAliveTime = System.currentTimeMillis();
 
-    private final List<Runnable> disconnectListener = new CopyOnWriteArrayList<>();
-
-    private final EmitterProcessor<TcpMessage> processor = EmitterProcessor.create(false);
-
-    private final FluxSink<TcpMessage> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
-
-    private final boolean serverClient;
+    public VertxTcpClient(String id, boolean serverClient) {
+        this.id = id;
+        this.serverClient = serverClient;
+    }
 
     @Override
     public void keepAlive() {
@@ -67,6 +64,43 @@ public class VertxTcpClient implements TcpClient {
         }
     }
 
+    @Override
+    public InetSocketAddress address() {
+        return getRemoteAddress();
+    }
+
+    @Override
+    public Mono<Void> sendMessage(EncodedMessage message) {
+        return Mono
+            .create((sink) -> {
+                if (socket == null) {
+                    sink.error(new SocketException("socket closed"));
+                    return;
+                }
+                Buffer buffer = Buffer.buffer(message.getPayload());
+                socket.write(buffer, r -> {
+                    keepAlive();
+                    if (r.succeeded()) {
+                        sink.success();
+                    } else {
+                        sink.error(r.cause());
+                    }
+                });
+            });
+    }
+
+    @Override
+    public Flux<EncodedMessage> receiveMessage() {
+        return this
+            .subscribe()
+            .cast(EncodedMessage.class);
+    }
+
+    @Override
+    public void disconnect() {
+        shutdown();
+    }
+
     @Override
     public boolean isAlive() {
         return socket != null && (keepAliveTimeoutMs < 0 || System.currentTimeMillis() - lastKeepAliveTime < keepAliveTimeoutMs);
@@ -77,11 +111,11 @@ public class VertxTcpClient implements TcpClient {
         return true;
     }
 
-    public VertxTcpClient(String id,boolean serverClient) {
-        this.id = id;
-        this.serverClient=serverClient;
-    }
-
+    /**
+     * 接收TCP消息
+     *
+     * @param message TCP消息
+     */
     protected void received(TcpMessage message) {
         if (processor.getPending() > processor.getBufferSize() / 2) {
             log.warn("tcp [{}] message pending {} ,drop message:{}", processor.getPending(), getRemoteAddress(), message.toString());
@@ -139,7 +173,7 @@ public class VertxTcpClient implements TcpClient {
             execute(runnable);
         }
         disconnectListener.clear();
-        if(serverClient){
+        if (serverClient) {
             processor.onComplete();
         }
     }
@@ -152,6 +186,11 @@ public class VertxTcpClient implements TcpClient {
         this.client = client;
     }
 
+    /**
+     * 设置客户端消息解析器
+     *
+     * @param payloadParser 消息解析器
+     */
     public void setRecordParser(PayloadParser payloadParser) {
         synchronized (this) {
             if (null != this.payloadParser && this.payloadParser != payloadParser) {
@@ -167,6 +206,11 @@ public class VertxTcpClient implements TcpClient {
         }
     }
 
+    /**
+     * socket处理
+     *
+     * @param socket socket
+     */
     public void setSocket(NetSocket socket) {
         synchronized (this) {
             Objects.requireNonNull(payloadParser);
@@ -193,21 +237,8 @@ public class VertxTcpClient implements TcpClient {
 
     @Override
     public Mono<Boolean> send(TcpMessage message) {
-        return Mono.<Boolean>create((sink) -> {
-            if (socket == null) {
-                sink.error(new SocketException("socket closed"));
-                return;
-            }
-            Buffer buffer = Buffer.buffer(message.getPayload());
-            socket.write(buffer, r -> {
-                keepAlive();
-                if (r.succeeded()) {
-                    sink.success(true);
-                } else {
-                    sink.error(r.cause());
-                }
-            });
-        });
+        return sendMessage(message)
+            .thenReturn(true);
     }
 
     @Override

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClientProvider.java

@@ -5,12 +5,11 @@ import io.vertx.core.net.NetClient;
 import io.vertx.core.net.NetClientOptions;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.bean.FastBeanCopier;
-import org.jetlinks.core.Values;
-import org.jetlinks.core.metadata.ConfigMetadata;
 import org.jetlinks.community.network.*;
 import org.jetlinks.community.network.security.CertificateManager;
 import org.jetlinks.community.network.security.VertxKeyCertTrustOptions;
 import org.jetlinks.community.network.tcp.parser.PayloadParserBuilder;
+import org.jetlinks.core.metadata.ConfigMetadata;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;
 

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

@@ -2,13 +2,13 @@ package org.jetlinks.community.network.tcp.device;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
+import org.jetlinks.community.network.tcp.TcpMessage;
+import org.jetlinks.community.network.tcp.client.TcpClient;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.session.DeviceSession;
-import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
-import org.jetlinks.community.network.tcp.TcpMessage;
-import org.jetlinks.community.network.tcp.client.TcpClient;
 import reactor.core.publisher.Mono;
 
 import java.net.InetSocketAddress;

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

@@ -3,8 +3,20 @@ 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;
+import org.jetlinks.core.device.DeviceProductOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.DeviceMessage;
 import org.jetlinks.core.message.Message;
@@ -12,18 +24,9 @@ import org.jetlinks.core.message.codec.DefaultTransport;
 import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.FromDeviceMessageContext;
 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;
@@ -45,6 +48,9 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
     @Getter
     private final String id;
 
+    /**
+     * 维护所有创建的tcp server
+     */
     private final TcpServer tcpServer;
 
     private final String protocol;
@@ -57,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);
@@ -64,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,
@@ -90,23 +100,91 @@ 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;
+        }
+        // 从TCPServer中获取连接的client
+        // client实例化为TcpConnection之后处理消息
+        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;
+    }
 
-    class TcpConnection {
+    @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<>();
         final AtomicReference<DeviceSession> sessionRef = new AtomicReference<>();
@@ -133,6 +211,7 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                     @Override
                     public void setKeepAliveTimeout(Duration timeout) {
                         keepaliveTimeout.set(timeout);
+                        client.setKeepAliveTimeout(timeout);
                     }
 
                     @Override
@@ -146,21 +225,36 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
 
         }
 
+        /**
+         * 接收消息
+         *
+         * @return void
+         */
         Mono<Void> accept() {
-            return client
-                .subscribe()
-                .filter(tcp -> started.get())
-                .publishOn(Schedulers.parallel())
-                .flatMap(this::handleTcpMessage)
-                .onErrorResume((err) -> {
-                    log.error(err.getMessage(), err);
-                    client.shutdown();
-                    return Mono.empty();
-                })
-                .then()
+            return getProtocol()
+                .flatMap(protocol -> protocol.onClientConnect(getTransport(), client, this))
+                .then(
+                    client
+                        .subscribe()
+                        .filter(tcp -> started.get())
+                        .publishOn(Schedulers.parallel())
+                        .flatMap(this::handleTcpMessage)
+                        .onErrorResume((err) -> {
+                            log.error(err.getMessage(), err);
+                            client.shutdown();
+                            return Mono.empty();
+                        })
+                        .then()
+                )
                 .doOnCancel(client::shutdown);
         }
 
+        /**
+         * 处理TCP消息 ==>> 设备消息
+         *
+         * @param message tcp消息
+         * @return void
+         */
         Mono<Void> handleTcpMessage(TcpMessage message) {
             return getProtocol()
                 .flatMap(pt -> pt.getMessageCodec(getTransport()))
@@ -168,77 +262,52 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                 .cast(DeviceMessage.class)
                 .doOnNext(msg -> gatewayMonitor.receivedMessage())
                 .flatMap(this::handleDeviceMessage)
-                .doOnEach(ReactiveLogger.onError(err -> log
-                    .error("处理TCP[{}]消息失败:\n{}", address, message, err)))
+                .doOnEach(ReactiveLogger.onError(err -> log.error("处理TCP[{}]消息失败:\n{}",
+                    address,
+                    message
+                    , err)))
                 .onErrorResume((err) -> Mono.fromRunnable(client::reset))
                 .then();
         }
 
+        /**
+         * 处理设备消息
+         *
+         * @param message 设备消息
+         * @return void
+         */
         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();
         }
 
-    }
-
-
-    private void doStart() {
-        if (started.getAndSet(true) || disposable != null) {
-            return;
+        @Override
+        public Mono<DeviceOperator> getDevice(String deviceId) {
+            return registry.getDevice(deviceId);
         }
-        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 Mono<DeviceProductOperator> getProduct(String productId) {
+            return registry.getProduct(productId);
+        }
 
-    @Override
-    public boolean isAlive() {
-        return started.get();
+        @Override
+        public Mono<Void> onMessage(DeviceMessage message) {
+            return handleDeviceMessage(message);
+        }
     }
 }

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

@@ -1,8 +1,5 @@
 package org.jetlinks.community.network.tcp.device;
 
-import org.jetlinks.core.ProtocolSupports;
-import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.community.gateway.DeviceGateway;
 import org.jetlinks.community.gateway.supports.DeviceGatewayProperties;
 import org.jetlinks.community.gateway.supports.DeviceGatewayProvider;
@@ -10,11 +7,20 @@ import org.jetlinks.community.network.DefaultNetworkType;
 import org.jetlinks.community.network.NetworkManager;
 import org.jetlinks.community.network.NetworkType;
 import org.jetlinks.community.network.tcp.server.TcpServer;
+import org.jetlinks.core.ProtocolSupports;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 import reactor.core.publisher.Mono;
 
+/**
+ * TCP服务设备网关提供商
+ *
+ * @author zhouhao
+ * @since 1.0
+ */
 @Component
 public class TcpServerDeviceGatewayProvider implements DeviceGatewayProvider {
 
@@ -60,18 +66,18 @@ public class TcpServerDeviceGatewayProvider implements DeviceGatewayProvider {
     public Mono<DeviceGateway> createDeviceGateway(DeviceGatewayProperties properties) {
         return networkManager
             .<TcpServer>getNetwork(getNetworkType(), properties.getNetworkId())
-            .map(mqttServer -> {
+            .map(server -> {
                 String protocol = (String) properties.getConfiguration().get("protocol");
 
-                Assert.hasText(protocol,"protocol can not be empty");
+                Assert.hasText(protocol, "protocol can not be empty");
 
-               return new TcpServerDeviceGateway(properties.getId(),
+                return new TcpServerDeviceGateway(properties.getId(),
                     protocol,
                     protocolSupports,
                     registry,
                     messageHandler,
                     sessionManager,
-                    mqttServer
+                    server
                 );
             });
     }

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

@@ -1,12 +1,12 @@
 package org.jetlinks.community.network.tcp.device;
 
 import lombok.Getter;
+import org.jetlinks.community.network.tcp.TcpMessage;
+import org.jetlinks.community.network.tcp.client.TcpClient;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.message.codec.EncodedMessage;
 import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.server.session.DeviceSession;
-import org.jetlinks.community.network.tcp.TcpMessage;
-import org.jetlinks.community.network.tcp.client.TcpClient;
 import reactor.core.publisher.Mono;
 
 class UnknownTcpDeviceSession implements DeviceSession {

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/parser/DefaultPayloadParserBuilder.java

@@ -1,7 +1,6 @@
 package org.jetlinks.community.network.tcp.parser;
 
 import org.jetlinks.community.ValueObject;
-import org.jetlinks.core.Values;
 import org.jetlinks.community.network.tcp.parser.strateies.DelimitedPayloadParserBuilder;
 import org.jetlinks.community.network.tcp.parser.strateies.DirectPayloadParserBuilder;
 import org.jetlinks.community.network.tcp.parser.strateies.FixLengthPayloadParserBuilder;

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/parser/PayloadParserBuilder.java

@@ -1,7 +1,6 @@
 package org.jetlinks.community.network.tcp.parser;
 
 import org.jetlinks.community.ValueObject;
-import org.jetlinks.core.Values;
 
 public interface PayloadParserBuilder {
 

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/parser/strateies/DelimitedPayloadParserBuilder.java

@@ -3,8 +3,6 @@ package org.jetlinks.community.network.tcp.parser.strateies;
 import io.vertx.core.parsetools.RecordParser;
 import org.apache.commons.lang.StringEscapeUtils;
 import org.jetlinks.community.ValueObject;
-import org.jetlinks.core.Value;
-import org.jetlinks.core.Values;
 import org.jetlinks.community.network.tcp.parser.PayloadParserType;
 
 public class DelimitedPayloadParserBuilder extends VertxPayloadParserBuilder {

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/parser/strateies/DirectPayloadParserBuilder.java

@@ -2,7 +2,6 @@ package org.jetlinks.community.network.tcp.parser.strateies;
 
 import lombok.SneakyThrows;
 import org.jetlinks.community.ValueObject;
-import org.jetlinks.core.Values;
 import org.jetlinks.community.network.tcp.parser.DirectRecordParser;
 import org.jetlinks.community.network.tcp.parser.PayloadParser;
 import org.jetlinks.community.network.tcp.parser.PayloadParserBuilderStrategy;

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/parser/strateies/FixLengthPayloadParserBuilder.java

@@ -2,8 +2,6 @@ package org.jetlinks.community.network.tcp.parser.strateies;
 
 import io.vertx.core.parsetools.RecordParser;
 import org.jetlinks.community.ValueObject;
-import org.jetlinks.core.Value;
-import org.jetlinks.core.Values;
 import org.jetlinks.community.network.tcp.parser.PayloadParserType;
 
 public class FixLengthPayloadParserBuilder extends VertxPayloadParserBuilder {