merge from master

.editorconfig

@@ -2,6 +2,7 @@ root = true
 
 [*]
 charset = utf-8
+end_of_line = lf
 
 [*.java]
 indent_style = space

.github/workflows/maven.yml

@@ -19,7 +19,7 @@ jobs:
         path: ~/.m2
         key: jetlinks-community-maven-repository
     - name: Build with Maven
-      run: mvn -B package && cd jetlinks-standalone && mvn docker:build
+      run: mvn -B package -Pbuild && cd jetlinks-standalone && mvn docker:build
     - name: Login Docker Repo
       run: echo "${{ secrets.ALIYUN_DOCKER_REPO_PWD }}" | docker login registry.cn-shenzhen.aliyuncs.com -u ${{ secrets.ALIYUN_DOCKER_REPO_USERNAME }} --password-stdin
     - name: Push Docker

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

@@ -6,6 +6,7 @@ import io.netty.buffer.Unpooled;
 import org.apache.commons.beanutils.BeanUtilsBean;
 import org.apache.commons.beanutils.Converter;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.http.MediaType;
 
 @Configuration
 public class CommonConfiguration {
@@ -26,6 +27,14 @@ public class CommonConfiguration {
                 return convert(aClass, JSON.toJSONBytes(o));
             }
         }, ByteBuf.class);
+
+        BeanUtilsBean.getInstance().getConvertUtils().register(new Converter() {
+            @Override
+            public <T> T convert(Class<T> aClass, Object o) {
+                return (T) MediaType.valueOf(String.valueOf(o));
+            }
+        }, MediaType.class);
+
     }
 
 }

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

@@ -52,8 +52,8 @@ public class ElasticSearchConfiguration {
         return new ElasticRestClient(client, client);
     }
 
-    @Bean
-    public RestHighLevelClient elasticsearchRestHighLevelClient(ElasticRestClient client) {
+    @Bean(destroyMethod = "close")
+    public RestHighLevelClient restHighLevelClient(ElasticRestClient client) {
         return client.getWriteClient();
     }
 

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

@@ -27,6 +27,7 @@ import org.jetlinks.community.elastic.search.index.ElasticSearchIndexManager;
 import org.jetlinks.community.elastic.search.utils.ElasticSearchConverter;
 import org.jetlinks.community.elastic.search.utils.ReactorActionListener;
 import org.reactivestreams.Publisher;
+import org.springframework.context.annotation.DependsOn;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.BufferOverflowStrategy;
 import reactor.core.publisher.Flux;
@@ -47,6 +48,7 @@ import java.util.stream.Collectors;
  **/
 @Service
 @Slf4j
+@DependsOn("restHighLevelClient")
 public class DefaultElasticSearchService implements ElasticSearchService {
 
     private final ElasticRestClient restClient;

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

@@ -15,8 +15,8 @@ import java.util.StringJoiner;
 import java.util.function.BiFunction;
 
 class ProxyMessageListener implements MessageListener {
-    private Class<?> paramType;
-    private Object target;
+    private final Class<?> paramType;
+    private final Object target;
 
     BiFunction<Object, Object, Object> proxy;
 
@@ -74,6 +74,9 @@ class ProxyMessageListener implements MessageListener {
 
         if (message.getMessage() instanceof EncodableMessage) {
             Object payload = ((EncodableMessage) message.getMessage()).getNativePayload();
+            if(paramType.isInstance(payload)){
+                return payload;
+            }
             return FastBeanCopier.DEFAULT_CONVERT.convert(payload, paramType, new Class[]{});
         }
         return message;

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

@@ -30,14 +30,14 @@ public class VertxTcpClient extends AbstractTcpClient {
     volatile PayloadParser payloadParser;
 
     @Getter
-    private String id;
+    private final String id;
 
     @Setter
     private long keepAliveTimeoutMs = Duration.ofMinutes(10).toMillis();
 
     private volatile long lastKeepAliveTime = System.currentTimeMillis();
 
-    private List<Runnable> disconnectListener = new CopyOnWriteArrayList<>();
+    private final List<Runnable> disconnectListener = new CopyOnWriteArrayList<>();
 
     @Override
     public void keepAlive() {

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

@@ -44,23 +44,31 @@ import java.util.function.Function;
 class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGateway {
 
     @Getter
-    private String id;
+    private final String id;
 
-    private TcpServer tcpServer;
+    private final TcpServer tcpServer;
 
-    private String protocol;
+    private final String protocol;
 
-    private ProtocolSupports supports;
+    private final ProtocolSupports supports;
 
-    private DeviceRegistry registry;
+    private final DeviceRegistry registry;
 
-    private DecodedClientMessageHandler clientMessageHandler;
+    private final DecodedClientMessageHandler clientMessageHandler;
 
-    private DeviceSessionManager sessionManager;
+    private final DeviceSessionManager sessionManager;
 
-    private DeviceGatewayMonitor gatewayMonitor;
+    private final DeviceGatewayMonitor gatewayMonitor;
 
-    private LongAdder counter = new LongAdder();
+    private final LongAdder counter = new LongAdder();
+
+    private final EmitterProcessor<Message> processor = EmitterProcessor.create(false);
+
+    private final FluxSink<Message> sink = processor.sink();
+
+    private final AtomicBoolean started = new AtomicBoolean();
+
+    private final List<Disposable> disposable = new CopyOnWriteArrayList<>();
 
     public TcpServerDeviceGateway(String id,
                                   String protocol,
@@ -80,12 +88,6 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
     }
 
 
-    private EmitterProcessor<Message> processor = EmitterProcessor.create(false);
-
-    private FluxSink<Message> sink = processor.sink();
-
-    private AtomicBoolean started = new AtomicBoolean();
-
     public Mono<ProtocolSupport> getProtocol() {
         return supports.getProtocol(protocol);
     }
@@ -105,8 +107,6 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
         return DefaultNetworkType.TCP_SERVER;
     }
 
-    private List<Disposable> disposable = new CopyOnWriteArrayList<>();
-
     private void doStart() {
         if (started.getAndSet(true) || !disposable.isEmpty()) {
             return;
@@ -127,7 +127,7 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                 AtomicReference<DeviceSession> sessionRef = new AtomicReference<>(sessionManager.getSession(client.getId()));
                 client.subscribe()
                     .filter(r -> started.get())
-                    .takeWhile(r -> disposable != null)
+                    .takeWhile(r -> !disposable.isEmpty())
                     .doOnNext(r -> {
                         log.debug("收到TCP报文:\n{}", r);
                         gatewayMonitor.receivedMessage();
@@ -165,11 +165,6 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                                 return getSession().getOperator();
                             }
                         }))
-                        .switchIfEmpty(Mono.fromRunnable(() ->
-                            log.warn("无法识别的TCP客户端[{}]消息:\n{}",
-                                clientAddr,
-                                tcpMessage
-                            )))
                         .cast(DeviceMessage.class)
                         .flatMap(message -> registry
                             .getDevice(message.getDeviceId())
@@ -182,20 +177,20 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                                 );
                             }))
                             .flatMap(device -> {
-                                DeviceSession fSession = sessionRef.get() == null ?
-                                    sessionManager.getSession(device.getDeviceId()) :
-                                    sessionRef.get();
+                                DeviceSession fSession = sessionManager.getSession(device.getDeviceId());
                                 //处理设备上线消息
                                 if (message instanceof DeviceOnlineMessage) {
                                     if (fSession == null) {
-                                        fSession = new TcpDeviceSession(client.getId(), device, client, getTransport()) {
+                                        boolean keepOnline = message.getHeader(Headers.keepOnline).orElse(false);
+                                        String sessionId = device.getDeviceId();
+                                        fSession = new TcpDeviceSession(sessionId, device, client, getTransport()) {
                                             @Override
                                             public Mono<Boolean> send(EncodedMessage encodedMessage) {
                                                 return super.send(encodedMessage).doOnSuccess(r -> gatewayMonitor.sentMessage());
                                             }
                                         };
-                                        //保持设备一直在线.（通过短连接上报数据的场景.可以让设备一直为在线状态）
-                                        if (message.getHeader(Headers.keepOnline).orElse(false)) {
+                                        //保持设备一直在线.（短连接上报数据的场景.可以让设备一直为在线状态）
+                                        if (keepOnline) {
                                             fSession = new KeepOnlineSession(fSession, Duration.ofMillis(-1));
                                         } else {
                                             client.onDisconnect(() -> sessionManager.unregister(device.getDeviceId()));

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

@@ -7,11 +7,16 @@ import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.jetlinks.community.network.DefaultNetworkType;
 import org.jetlinks.community.network.NetworkType;
+import org.jetlinks.community.network.tcp.client.TcpClient;
 import org.jetlinks.community.network.tcp.client.VertxTcpClient;
 import org.jetlinks.community.network.tcp.parser.PayloadParser;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.function.Function;
 import java.util.function.Supplier;
 
 /**
@@ -19,7 +24,7 @@ import java.util.function.Supplier;
  * @since 1.0
  **/
 @Slf4j
-public class VertxTcpServer extends AbstractTcpServer implements TcpServer {
+public class VertxTcpServer implements TcpServer {
 
     Collection<NetServer> tcpServers;
 
@@ -29,12 +34,22 @@ public class VertxTcpServer extends AbstractTcpServer implements TcpServer {
     private long keepAliveTimeout = Duration.ofMinutes(10).toMillis();
 
     @Getter
-    private String id;
+    private final String id;
+
+    private final EmitterProcessor<TcpClient> processor = EmitterProcessor.create(false);
+
+    private final FluxSink<TcpClient> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
 
     public VertxTcpServer(String id) {
         this.id = id;
     }
 
+    @Override
+    public Flux<TcpClient> handleConnection() {
+        return processor
+            .map(Function.identity());
+    }
+
     private void execute(Runnable runnable) {
         try {
             runnable.run();
@@ -61,6 +76,11 @@ public class VertxTcpServer extends AbstractTcpServer implements TcpServer {
 
 
     protected void acceptTcpConnection(NetSocket socket) {
+        if (!processor.hasDownstreams()) {
+            log.warn("not handler for tcp client[{}]", socket.remoteAddress());
+            socket.close();
+            return;
+        }
         VertxTcpClient client = new VertxTcpClient(id + "_" + socket.remoteAddress());
         client.setKeepAliveTimeoutMs(keepAliveTimeout);
         try {
@@ -72,7 +92,7 @@ public class VertxTcpServer extends AbstractTcpServer implements TcpServer {
             });
             client.setRecordParser(parserSupplier.get());
             client.setSocket(socket);
-            received(client);
+            sink.next(client);
             log.debug("accept tcp client [{}] connection", socket.remoteAddress());
         } catch (Exception e) {
             log.error("create tcp server client error", e);

jetlinks-components/notify-component/notify-core/src/main/java/org/jetlinks/community/notify/event/SerializableNotifierEvent.java

@@ -1,8 +1,6 @@
 package org.jetlinks.community.notify.event;
 
-import lombok.Builder;
-import lombok.Getter;
-import lombok.Setter;
+import lombok.*;
 import org.jetlinks.community.notify.template.Template;
 
 import javax.annotation.Nonnull;
@@ -12,6 +10,8 @@ import java.util.Map;
 @Getter
 @Setter
 @Builder
+@NoArgsConstructor
+@AllArgsConstructor
 public class SerializableNotifierEvent {
 
     private boolean success;

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/JetLinksApplication.java

@@ -8,6 +8,7 @@ import org.hswebframework.web.loggin.aop.EnableAccessLogger;
 import org.hswebframework.web.logging.events.AccessLoggerAfterEvent;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.elasticsearch.rest.RestClientAutoConfiguration;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.Profile;
@@ -18,7 +19,8 @@ import javax.annotation.PostConstruct;
 
 
 @SpringBootApplication(scanBasePackages = "org.jetlinks.community", exclude = {
-    DataSourceAutoConfiguration.class
+    DataSourceAutoConfiguration.class,
+    RestClientAutoConfiguration.class
 })
 @EnableCaching
 @EnableEasyormRepository("org.jetlinks.community.**.entity")

pom.xml

@@ -29,6 +29,7 @@
         <elasticsearch.version>6.8.6</elasticsearch.version>
         <reactor.excel.version>1.0-RC</reactor.excel.version>
         <reactor.ql.version>1.0.0</reactor.ql.version>
+        <fastjson.version>1.2.70</fastjson.version>
     </properties>
 
     <build>
@@ -150,8 +151,22 @@
                 <version>3.0.0</version>
             </plugin>
         </plugins>
+
     </build>
 
+    <profiles>
+        <profile>
+            <id>build</id>
+            <repositories>
+                <repository>
+                    <id>maven-central</id>
+                    <name>central</name>
+                    <url>https://repo1.maven.org/maven2/</url>
+                </repository>
+            </repositories>
+        </profile>
+    </profiles>
+
     <dependencyManagement>
 
         <dependencies>
@@ -182,7 +197,7 @@
             <dependency>
                 <groupId>com.alibaba</groupId>
                 <artifactId>fastjson</artifactId>
-                <version>1.2.56</version>
+                <version>${fastjson.version}</version>
             </dependency>
 
             <dependency>