优化设备网关

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

@@ -25,6 +25,11 @@ public interface PropertyConstants {
 
     interface Key<V> extends ConfigKey<V>, HeaderKey<V> {
 
+        @Override
+        default Class<V> getType() {
+            return ConfigKey.super.getType();
+        }
+
         static <T> Key<T> of(String key) {
             return new Key<T>() {
                 @Override

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

@@ -2,29 +2,26 @@ 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.core.ProtocolSupport;
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.message.DeviceOfflineMessage;
-import org.jetlinks.core.message.DeviceOnlineMessage;
 import org.jetlinks.core.message.Message;
 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.MessageHandler;
-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.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;
@@ -32,7 +29,6 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 
-import javax.annotation.Nonnull;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
@@ -40,7 +36,6 @@ import java.util.Objects;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Function;
 
 @Slf4j
 public class MqttClientDeviceGateway implements DeviceGateway {
@@ -58,11 +53,9 @@ public class MqttClientDeviceGateway implements DeviceGateway {
 
     private final ProtocolSupports protocolSupport;
 
-    private final DecodedClientMessageHandler clientMessageHandler;
-
-    private final EmitterProcessor<Message> messageProcessor = EmitterProcessor.create(false);
+    private final EmitterProcessor<Message> processor = EmitterProcessor.create(false);
 
-    private final FluxSink<Message> sink = messageProcessor.sink(FluxSink.OverflowStrategy.BUFFER);
+    private final FluxSink<Message> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
 
     private final AtomicBoolean started = new AtomicBoolean();
 
@@ -70,7 +63,7 @@ public class MqttClientDeviceGateway implements DeviceGateway {
 
     private final DeviceGatewayMonitor gatewayMonitor;
 
-    private final DeviceSessionManager sessionManager;
+    private final DeviceGatewayHelper helper;
 
     public MqttClientDeviceGateway(String id,
                                    MqttClient mqttClient,
@@ -87,9 +80,8 @@ public class MqttClientDeviceGateway implements DeviceGateway {
         this.registry = Objects.requireNonNull(registry, "registry");
         this.protocolSupport = Objects.requireNonNull(protocolSupport, "protocolSupport");
         this.protocol = Objects.requireNonNull(protocol, "protocol");
-        this.clientMessageHandler = Objects.requireNonNull(clientMessageHandler, "clientMessageHandler");
-        this.sessionManager = Objects.requireNonNull(sessionManager, "sessionManager");
         this.topics = Objects.requireNonNull(topics, "topics");
+        this.helper = new DeviceGatewayHelper(registry, sessionManager, clientMessageHandler);
     }
 
 
@@ -101,86 +93,62 @@ public class MqttClientDeviceGateway implements DeviceGateway {
         if (started.getAndSet(true) || !disposable.isEmpty()) {
             return;
         }
-        disposable.add(mqttClient
-            .subscribe(topics)
-            .filter((msg) -> started.get())
-            .flatMap(mqttMessage -> {
-                AtomicReference<Duration> timeoutRef = new AtomicReference<>();
-                return getProtocol()
-                    .flatMap(codec -> codec.getMessageCodec(getTransport()))
-                    .flatMapMany(codec -> codec.decode(new FromDeviceMessageContext() {
-                        @Nonnull
-                        @Override
-                        public EncodedMessage getMessage() {
-                            return mqttMessage;
-                        }
-
-                        @Override
-                        public DeviceSession getSession() {
-                            return new UnknownDeviceMqttClientSession(id + ":unknown", mqttClient) {
-                                @Override
-                                public Mono<Boolean> send(EncodedMessage encodedMessage) {
-                                    return super.send(encodedMessage).doOnSuccess(r -> gatewayMonitor.sentMessage());
-                                }
-
-                                @Override
-                                public void setKeepAliveTimeout(Duration timeout) {
-                                    timeoutRef.set(timeout);
-                                }
-                            };
-                        }
-
-                        @Override
-                        public DeviceOperator getDevice() {
-                            return null;
-                        }
-
-                        @Override
-                        public Mono<DeviceOperator> getDevice(String deviceId) {
-                            return registry.getDevice(deviceId);
-                        }
-                    }))
-                    .doOnError((err) -> log.error("解码MQTT客户端消息失败 {}:{}",
-                        mqttMessage.getTopic(), mqttMessage.getPayload().toString(StandardCharsets.UTF_8), err))
-                    .cast(DeviceMessage.class)
-                    .flatMap(msg -> {
-                        gatewayMonitor.receivedMessage();
-                        if (messageProcessor.hasDownstreams()) {
-                            sink.next(msg);
-                        }
-                        return registry
-                            .getDevice(msg.getDeviceId())
-                            .switchIfEmpty(Mono.fromRunnable(() -> log.debug("无法识别的设备:{}", msg)))
-                            .flatMap(device -> {
-                                DeviceSession session = sessionManager.getSession(device.getDeviceId());
-                                if (session == null) {
-                                    session = new MqttClientSession(id + ":" + device.getDeviceId(), device, mqttClient) {
-                                        @Override
-                                        public Mono<Boolean> send(EncodedMessage encodedMessage) {
-                                            return super.send(encodedMessage).doOnSuccess(r -> gatewayMonitor.sentMessage());
-                                        }
-                                    };
-                                    if (!(msg instanceof DeviceOfflineMessage)) {
-                                        sessionManager.register(session);
-                                    }
-                                }
-                                session.keepAlive();
-                                if (timeoutRef.get() != null) {
-                                    session.setKeepAliveTimeout(timeoutRef.get());
-                                }
-                                if (msg instanceof DeviceOnlineMessage) {
-                                    return Mono.empty();
-                                } else if (msg instanceof DeviceOfflineMessage) {
-                                    return Mono.fromRunnable(() -> sessionManager.unregister(device.getDeviceId()));
-                                } else {
-                                    return clientMessageHandler.handleMessage(device, msg).then();
-                                }
-                            });
-                    })
-                    .onErrorContinue((err, ms) -> log.error("处理MQTT消息失败:{}", mqttMessage, err));
-            })
-            .onErrorContinue((err, ms) -> log.error("处理MQTT客户端消息失败", err))
-            .subscribe());
+        disposable
+            .add(mqttClient
+                     .subscribe(topics)
+                     .filter((msg) -> started.get())
+                     .flatMap(mqttMessage -> {
+                         AtomicReference<Duration> timeoutRef = new AtomicReference<>();
+                         return this
+                             .getProtocol()
+                             .flatMap(codec -> codec.getMessageCodec(getTransport()))
+                             .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(
+                                 new UnknownDeviceMqttClientSession(id + ":unknown", mqttClient) {
+                                     @Override
+                                     public Mono<Boolean> send(EncodedMessage encodedMessage) {
+                                         return super
+                                             .send(encodedMessage)
+                                             .doOnSuccess(r -> gatewayMonitor.sentMessage());
+                                     }
+
+                                     @Override
+                                     public void setKeepAliveTimeout(Duration timeout) {
+                                         timeoutRef.set(timeout);
+                                     }
+                                 }
+                                 , mqttMessage, registry)
+                             ))
+                             .doOnError((err) -> log.error("解码MQTT客户端消息失败 {}:{}",
+                                                           mqttMessage.getTopic(),
+                                                           mqttMessage
+                                                               .getPayload()
+                                                               .toString(StandardCharsets.UTF_8),
+                                                           err))
+                             .cast(DeviceMessage.class)
+                             .flatMap(message -> {
+                                 if (processor.hasDownstreams()) {
+                                     sink.next(message);
+                                 }
+                                 gatewayMonitor.receivedMessage();
+                                 return helper
+                                     .handleDeviceMessage(message,
+                                                          device -> createDeviceSession(device, mqttClient),
+                                                          DeviceGatewayHelper.applySessionKeepaliveTimeout(message, timeoutRef::get),
+                                                          () -> log.warn("无法从MQTT[{}]消息中获取设备信息:{}", mqttMessage.print(), message)
+                                     );
+                             })
+                             .then()
+                             .onErrorResume((err) -> {
+                                 log.error("处理MQTT消息失败:{}", mqttMessage, err);
+                                 return Mono.empty();
+                             });
+                     }, Integer.MAX_VALUE)
+                     .onErrorContinue((err, ms) -> log.error("处理MQTT客户端消息失败", err))
+                     .subscribe());
+    }
+
+    private MqttClientSession createDeviceSession(DeviceOperator device, MqttClient client) {
+        return new MqttClientSession(device.getDeviceId(), device, client, gatewayMonitor);
     }
 
     @Override
@@ -195,7 +163,7 @@ public class MqttClientDeviceGateway implements DeviceGateway {
 
     @Override
     public Flux<Message> onMessage() {
-        return messageProcessor.map(Function.identity());
+        return processor;
     }
 
     @Override

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

@@ -2,7 +2,7 @@ package org.jetlinks.community.network.mqtt.gateway.device;
 
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.server.MessageHandler;
+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,7 +10,6 @@ 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.server.session.DeviceSessionManager;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;
@@ -20,7 +19,6 @@ import java.util.Objects;
 
 @Component
 public class MqttClientDeviceGatewayProvider implements DeviceGatewayProvider {
-
     private final NetworkManager networkManager;
 
     private final DeviceRegistry registry;
@@ -50,7 +48,7 @@ public class MqttClientDeviceGatewayProvider implements DeviceGatewayProvider {
 
     @Override
     public String getName() {
-        return "MQTT客户端设备网关";
+        return "MQTT Broker接入";
     }
 
     @Override
@@ -69,13 +67,13 @@ public class MqttClientDeviceGatewayProvider implements DeviceGatewayProvider {
                 Objects.requireNonNull(topics, "topics");
 
                 return new MqttClientDeviceGateway(properties.getId(),
-                    mqttClient,
-                    registry,
-                    protocolSupports,
-                    protocol,
-                    sessionManager,
-                    clientMessageHandler,
-                    Arrays.asList(topics.split("[,;\n]"))
+                                                   mqttClient,
+                                                   registry,
+                                                   protocolSupports,
+                                                   protocol,
+                                                   sessionManager,
+                                                   clientMessageHandler,
+                                                   Arrays.asList(topics.split("[,;\n]"))
                 );
 
             });

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

@@ -4,26 +4,32 @@ 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.monitor.DeviceGatewayMonitor;
-import org.jetlinks.community.gateway.monitor.GatewayMonitors;
-import org.jetlinks.community.gateway.monitor.MonitorSupportDeviceGateway;
 import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.device.AuthenticationResponse;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.device.MqttAuthenticationRequest;
-import org.jetlinks.core.message.*;
-import org.jetlinks.core.message.codec.*;
+import org.jetlinks.core.message.CommonDeviceMessage;
+import org.jetlinks.core.message.CommonDeviceMessageReply;
+import org.jetlinks.core.message.DeviceMessage;
+import org.jetlinks.core.message.Message;
+import org.jetlinks.core.message.codec.DefaultTransport;
+import org.jetlinks.core.message.codec.FromDeviceMessageContext;
+import org.jetlinks.core.message.codec.MqttMessage;
+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.core.server.session.KeepOnlineSession;
-import org.jetlinks.core.server.session.ReplaceableDeviceSession;
+import org.jetlinks.community.network.utils.DeviceGatewayHelper;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import org.springframework.util.StringUtils;
 import reactor.core.Disposable;
@@ -35,8 +41,6 @@ import reactor.core.scheduler.Schedulers;
 import reactor.util.function.Tuple3;
 import reactor.util.function.Tuples;
 
-import javax.annotation.Nonnull;
-import java.time.Duration;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
@@ -69,13 +73,14 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
 
     private Disposable disposable;
 
+    private final DeviceGatewayHelper helper;
+
     public MqttServerDeviceGateway(String id,
                                    DeviceRegistry registry,
                                    DeviceSessionManager sessionManager,
                                    MqttServer mqttServer,
                                    DecodedClientMessageHandler messageHandler,
-                                   Mono<ProtocolSupport> customProtocol
-    ) {
+                                   Mono<ProtocolSupport> customProtocol) {
         this.gatewayMonitor = GatewayMonitors.getDeviceGatewayMonitor(id);
         this.id = id;
         this.registry = registry;
@@ -83,6 +88,7 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
         this.mqttServer = mqttServer;
         this.messageHandler = messageHandler;
         this.supportMono = customProtocol;
+        this.helper = new DeviceGatewayHelper(registry, sessionManager, messageHandler);
     }
 
     @Override
@@ -115,11 +121,11 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
 
     //处理连接，并进行认证
     private Mono<Tuple3<DeviceOperator, AuthenticationResponse, MqttConnection>> handleConnection(MqttConnection connection) {
-
         return Mono
             .justOrEmpty(connection.getAuth())
             .flatMap(auth -> {
-                MqttAuthenticationRequest request = new MqttAuthenticationRequest(connection.getClientId(), auth.getUsername(), auth.getPassword(), getTransport());
+                MqttAuthenticationRequest request = new MqttAuthenticationRequest(connection.getClientId(), auth.getUsername(), auth
+                    .getPassword(), getTransport());
                 return supportMono
                     //使用自定义协议来认证
                     .map(support -> support.authenticate(request, registry))
@@ -175,7 +181,11 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
                         gatewayMonitor.disconnected();
                         gatewayMonitor.totalConnection(counter.sum());
                     });
-                    return Tuples.of(connection.accept(), device, newSession);
+                    try {
+                        return Tuples.of(connection.accept(), device, newSession);
+                    } catch (IllegalStateException ignore) {
+
+                    }
                 } else {
                     connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                     gatewayMonitor.rejected();
@@ -203,9 +213,9 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
             .publishOn(Schedulers.parallel())
             .doOnNext(msg -> gatewayMonitor.receivedMessage())
             .flatMap(publishing ->
-                this.decodeAndHandleMessage(operator, session, publishing.getMessage(), connection)
-                    //ack
-                    .doOnSuccess(s -> publishing.acknowledge())
+                         this.decodeAndHandleMessage(operator, session, publishing.getMessage(), connection)
+                             //ack
+                             .doOnSuccess(s -> publishing.acknowledge())
             )
             //合并遗言消息
             .mergeWith(
@@ -224,30 +234,25 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
         return operator
             .getProtocol()
             .flatMap(protocol -> protocol.getMessageCodec(getTransport()))
-            .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(session, message,registry)))
+            .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(session, message, registry)))
             .cast(DeviceMessage.class)
             .flatMap(msg -> {
+                if (messageProcessor.hasDownstreams()) {
+                    sink.next(msg);
+                }
                 if (msg instanceof CommonDeviceMessage) {
                     CommonDeviceMessage _msg = ((CommonDeviceMessage) msg);
                     if (StringUtils.isEmpty(_msg.getDeviceId())) {
                         _msg.setDeviceId(operator.getDeviceId());
                     }
                 }
-                String deviceId = msg.getDeviceId();
-                if (!StringUtils.isEmpty(deviceId)) {
-                    //返回了其他设备的消息,则自动创建会话
-                    if (!deviceId.equals(operator.getDeviceId())) {
-                        DeviceSession anotherSession = sessionManager.getSession(msg.getDeviceId());
-                        if (anotherSession == null) {
-                            return registry
-                                .getDevice(msg.getDeviceId())
-                                .map(device -> handleMessage(device.getDeviceId(), device, msg, session))
-                                .defaultIfEmpty(Mono.defer(()->handleMessage(msg.getDeviceId(), operator, msg, session)))
-                                .flatMap(Function.identity());
-                        }
+                if (msg instanceof CommonDeviceMessageReply) {
+                    CommonDeviceMessageReply<?> _msg = ((CommonDeviceMessageReply<?>) msg);
+                    if (StringUtils.isEmpty(_msg.getDeviceId())) {
+                        _msg.setDeviceId(operator.getDeviceId());
                     }
                 }
-                return handleMessage(deviceId, operator, msg, session);
+                return handleMessage(operator, msg, connection);
             })
             .then()
             .doOnEach(ReactiveLogger.onError(err -> log.error("处理MQTT连接[{}]消息失败:{}", operator.getDeviceId(), message, err)))
@@ -255,76 +260,27 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
             ;
     }
 
-    protected Mono<Boolean> handleOnlineOffline(DeviceMessage message, MqttConnectionSession firstSession) {
-        if (message == null || message.getDeviceId() == null) {
-            return Mono.just(false);
-        }
-        String deviceId = message.getDeviceId();
-        Mono<Boolean> then = Mono.empty();
-        if (message instanceof ChildDeviceMessage) {
-            then = handleOnlineOffline((DeviceMessage) ((ChildDeviceMessage) message).getChildDeviceMessage(), firstSession);
-        } else if (message instanceof ChildDeviceMessageReply) {
-            then = handleOnlineOffline((DeviceMessage) ((ChildDeviceMessageReply) message).getChildDeviceMessage(), firstSession);
-        } else if (message instanceof DeviceOnlineMessage) {
-            then = Mono.just(true);
-        } else if (message instanceof DeviceOfflineMessage) {
-            sessionManager.unregister(deviceId);
-            return Mono.just(true);
-        }
-        if (firstSession.isAlive()) {
-            DeviceSession[] managedSession = new DeviceSession[]{
-                sessionManager.getSession(deviceId)
-            };
-            //session 不存在,可能是同一个mqtt返回多个设备消息
-            if (managedSession[0] == null) {
-                then = registry
-                    .getDevice(deviceId)
-                    .doOnNext(device -> sessionManager
-                        .register(managedSession[0] =
-                                      new MqttConnectionSession(deviceId,
-                                                                device,
-                                                                getTransport(),
-                                                                firstSession.getConnection(),
-                                                                gatewayMonitor)))
-                    .then(then);
-            } else {
-                managedSession[0] = firstSession;
-            }
-
-            //保持会话，在低功率设备上,可能无法保持mqtt长连接.
-            if (message.getHeader(Headers.keepOnline).orElse(false)) {
-                then = Mono
-                    .fromRunnable(() -> {
-                        DeviceSession session = managedSession[0];
-                        if (!session.isWrapFrom(KeepOnlineSession.class)) {
-                            int timeout = message.getHeaderOrDefault(Headers.keepOnlineTimeoutSeconds);
-                            KeepOnlineSession keepOnlineSession = new KeepOnlineSession(session, Duration.ofSeconds(timeout));
-                            //替换会话
-                            session = sessionManager.replace(session, keepOnlineSession);
-                            session.keepAlive();
-                        }
-                    })
-                    .then(then);
-            }else if(managedSession[0]!=null) {
-                managedSession[0].keepAlive();
-            }
-        }
-
-        return then;
-    }
-
-    private Mono<Void> handleMessage(String deviceId,
-                                     DeviceOperator device,
+    private Mono<Void> handleMessage(DeviceOperator mainDevice,
                                      DeviceMessage message,
-                                     MqttConnectionSession firstSession) {
-        if (messageProcessor.hasDownstreams()) {
-            sink.next(message);
+                                     MqttConnection connection) {
+        if (!connection.isAlive()) {
+            return messageHandler
+                .handleMessage(mainDevice, message)
+                .then();
         }
-
-        return handleOnlineOffline(message, firstSession)
-            //只有empty才转发消息
-            .switchIfEmpty(messageHandler.handleMessage(device, message))
-            .then();
+        return helper.handleDeviceMessage(message,
+                                          device -> new MqttConnectionSession(device.getDeviceId(),
+                                                                              device,
+                                                                              getTransport(),
+                                                                              connection,
+                                                                              gatewayMonitor),
+                                          session -> {
+
+                                          },
+                                          () -> {
+                                              log.warn("无法从MQTT[{}]消息中获取设备信息:{}", connection.getClientId(), message);
+                                          })
+                     .then();
     }
 
     @Override
@@ -339,8 +295,7 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
 
     @Override
     public Flux<Message> onMessage() {
-        return messageProcessor
-            .map(Function.identity());
+        return messageProcessor;
     }
 
     @Override

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

@@ -36,7 +36,7 @@ public class MqttServerDeviceGatewayProvider implements DeviceGatewayProvider {
         this.registry = registry;
         this.sessionManager = sessionManager;
         this.messageHandler = messageHandler;
-        this.protocolSupports=protocolSupports;
+        this.protocolSupports = protocolSupports;
     }
 
     @Override
@@ -46,7 +46,7 @@ public class MqttServerDeviceGatewayProvider implements DeviceGatewayProvider {
 
     @Override
     public String getName() {
-        return "MQTT服务设备网关";
+        return "MQTT直连接入";
     }
 
     @Override
@@ -56,6 +56,7 @@ public class MqttServerDeviceGatewayProvider implements DeviceGatewayProvider {
 
     @Override
     public Mono<DeviceGateway> createDeviceGateway(DeviceGatewayProperties properties) {
+
         return networkManager
             .<MqttServer>getNetwork(getNetworkType(), properties.getNetworkId())
             .map(mqttServer -> new MqttServerDeviceGateway(

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

@@ -1,12 +1,14 @@
 package org.jetlinks.community.network.mqtt.gateway.device.session;
 
 import lombok.Getter;
+import lombok.Setter;
 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;
 
@@ -14,25 +16,31 @@ import java.time.Duration;
 
 public class MqttClientSession implements DeviceSession {
     @Getter
-    private String id;
+    private final String id;
 
     @Getter
-    private DeviceOperator operator;
+    private final DeviceOperator operator;
 
+    @Getter
+    @Setter
     private MqttClient client;
 
-    private long connectTime = System.currentTimeMillis();
+    private final long connectTime = System.currentTimeMillis();
 
     private long lastPingTime = System.currentTimeMillis();
 
     private long keepAliveTimeout = -1;
 
+    private final DeviceGatewayMonitor monitor;
+
     public MqttClientSession(String id,
                              DeviceOperator operator,
-                             MqttClient client) {
+                             MqttClient client,
+                             DeviceGatewayMonitor monitor) {
         this.id = id;
         this.operator = operator;
         this.client = client;
+        this.monitor=monitor;
     }
 
     @Override
@@ -53,8 +61,11 @@ public class MqttClientSession implements DeviceSession {
     @Override
     public Mono<Boolean> send(EncodedMessage encodedMessage) {
         if (encodedMessage instanceof MqttMessage) {
-            return client.publish(((MqttMessage) encodedMessage))
-                .thenReturn(true);
+            monitor.sentMessage();
+            return client
+                .publish(((MqttMessage) encodedMessage))
+                .thenReturn(true)
+                ;
         }
         return Mono.error(new UnsupportedOperationException("unsupported message type:" + encodedMessage.getClass()));
     }

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

@@ -40,5 +40,12 @@
             <artifactId>vertx-core</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.jetlinks.community</groupId>
+            <artifactId>common-component</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+
     </dependencies>
 </project>

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

@@ -4,7 +4,9 @@ package org.jetlinks.community.network.utils;
  * 字节转换工具
  *
  * @author zhouhao
+ * @see org.jetlinks.core.utils.BytesUtils
  */
+@Deprecated
 public class BytesUtils {
 
     /**

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

@@ -0,0 +1,138 @@
+package org.jetlinks.community.network.utils;
+
+import lombok.AllArgsConstructor;
+import org.jetlinks.community.PropertyConstants;
+import org.jetlinks.core.device.DeviceOperator;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.core.message.*;
+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 reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+@AllArgsConstructor
+public class DeviceGatewayHelper {
+
+    private final DeviceRegistry registry;
+    private final DeviceSessionManager sessionManager;
+    private final DecodedClientMessageHandler messageHandler;
+
+
+    public static Consumer<DeviceSession> applySessionKeepaliveTimeout(DeviceMessage msg, Supplier<Duration> timeoutSupplier) {
+        return session -> {
+            Duration timeout = msg
+                .getHeader(Headers.keepOnlineTimeoutSeconds)
+                .map(Duration::ofSeconds)
+                .orElseGet(timeoutSupplier);
+            if (null != timeout) {
+                session.setKeepAliveTimeout(timeout);
+            }
+        };
+    }
+
+    public Mono<DeviceOperator> handleDeviceMessage(DeviceMessage message,
+                                                    Function<DeviceOperator, DeviceSession> sessionBuilder,
+                                                    Consumer<DeviceSession> sessionConsumer,
+                                                    Runnable deviceNotFoundListener) {
+
+        return handleDeviceMessage(message, sessionBuilder, sessionConsumer, () -> Mono.fromRunnable(deviceNotFoundListener));
+    }
+
+    public Mono<DeviceOperator> handleDeviceMessage(DeviceMessage message,
+                                                    Function<DeviceOperator, DeviceSession> sessionBuilder,
+                                                    Consumer<DeviceSession> sessionConsumer,
+                                                    Supplier<Mono<DeviceOperator>> deviceNotFoundListener) {
+
+        return handleDeviceMessage(message, sessionBuilder, sessionConsumer, deviceNotFoundListener, false);
+    }
+
+    private Mono<DeviceOperator> handleDeviceMessage(DeviceMessage message,
+                                                     Function<DeviceOperator, DeviceSession> sessionBuilder,
+                                                     Consumer<DeviceSession> sessionConsumer,
+                                                     Supplier<Mono<DeviceOperator>> deviceNotFoundListener,
+                                                     boolean children) {
+        String deviceId = message.getDeviceId();
+        Mono<DeviceOperator> then = Mono.empty();
+        if (message instanceof ChildDeviceMessage) {
+            then = handleDeviceMessage((DeviceMessage) ((ChildDeviceMessage) message).getChildDeviceMessage(),
+                                       sessionBuilder,
+                                       sessionConsumer,
+                                       deviceNotFoundListener, true);
+        } else if (message instanceof ChildDeviceMessageReply) {
+            then = handleDeviceMessage((DeviceMessage) ((ChildDeviceMessageReply) message).getChildDeviceMessage(),
+                                       sessionBuilder,
+                                       sessionConsumer,
+                                       deviceNotFoundListener, true);
+        } else if (message instanceof DeviceOnlineMessage) {
+            //设备在线消息
+            then = registry.getDevice(deviceId);
+        } else if (message instanceof DeviceOfflineMessage) {
+            //设备离线消息
+            DeviceSession session = sessionManager.unregister(deviceId);
+            if (null == session) {
+                //如果session不存在,则将离线消息转发到
+                return registry
+                    .getDevice(deviceId)
+                    .flatMap(device -> messageHandler
+                        .handleMessage(device, message)
+                        .thenReturn(device));
+            }
+            return registry.getDevice(deviceId);
+        }
+        DeviceSession session = sessionManager.getSession(deviceId);
+        //session不存在,可能是同一个连接返回多个设备消息
+        if (session == null) {
+            return registry
+                .getDevice(deviceId)
+                .switchIfEmpty(children ? Mono.empty() : Mono.defer(() -> {
+                    //设备注册
+                    if (message instanceof DeviceRegisterMessage) {
+                        if (message.getHeader(PropertyConstants.deviceName).isPresent()
+                            && message.getHeader(PropertyConstants.productId).isPresent()) {
+                            return messageHandler
+                                .handleMessage(null, message)
+                                //延迟2秒后尝试重新获取设备并上线
+                                .then(Mono.delay(Duration.ofSeconds(2)))
+                                .then(registry.getDevice(deviceId));
+                        }
+                    }
+                    if (deviceNotFoundListener != null) {
+                        return deviceNotFoundListener.get();
+                    }
+                    return Mono.empty();
+                }))
+                .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) {
+                        return messageHandler
+                            .handleMessage(device, message)
+                            .thenReturn(device);
+                    }
+                    return Mono.just(device);
+                })
+                .then(then);
+        } else {
+            sessionConsumer.accept(session);
+            session.keepAlive();
+            return then
+                .then(messageHandler.handleMessage(session.getOperator(), message))
+                .then(registry.getDevice(deviceId));
+        }
+
+    }
+
+}

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

@@ -6,18 +6,18 @@ 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;
 import java.time.Duration;
+import java.util.Objects;
 import java.util.Optional;
 
 class TcpDeviceSession implements DeviceSession {
 
-    @Getter
-    private String id;
 
     @Getter
     @Setter
@@ -27,17 +27,27 @@ class TcpDeviceSession implements DeviceSession {
     private TcpClient client;
 
     @Getter
-    private Transport transport;
+    private final Transport transport;
 
     private long lastPingTime = System.currentTimeMillis();
 
-    private long connectTime = System.currentTimeMillis();
+    private final long connectTime = System.currentTimeMillis();
+
+    private final DeviceGatewayMonitor monitor;
 
-    TcpDeviceSession(String id, DeviceOperator operator, TcpClient client, Transport transport) {
-        this.id = id;
+    TcpDeviceSession(DeviceOperator operator,
+                     TcpClient client,
+                     Transport transport,
+                     DeviceGatewayMonitor monitor) {
         this.operator = operator;
         this.client = client;
         this.transport = transport;
+        this.monitor=monitor;
+    }
+
+    @Override
+    public String getId() {
+        return getDeviceId();
     }
 
     @Override
@@ -57,6 +67,7 @@ class TcpDeviceSession implements DeviceSession {
 
     @Override
     public Mono<Boolean> send(EncodedMessage encodedMessage) {
+        monitor.sentMessage();
         return client.send(new TcpMessage(encodedMessage.getPayload()));
     }
 
@@ -71,7 +82,6 @@ class TcpDeviceSession implements DeviceSession {
         client.keepAlive();
     }
 
-
     @Override
     public void setKeepAliveTimeout(Duration timeout) {
         client.setKeepAliveTimeout(timeout);
@@ -91,4 +101,17 @@ class TcpDeviceSession implements DeviceSession {
     public void onClose(Runnable call) {
         client.onDisconnect(call);
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        TcpDeviceSession session = (TcpDeviceSession) o;
+        return Objects.equals(client, session.client);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(client);
+    }
 }

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

@@ -3,26 +3,27 @@ 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.core.ProtocolSupport;
 import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.*;
+import org.jetlinks.core.message.DeviceMessage;
+import org.jetlinks.core.message.Message;
 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.session.DeviceSession;
 import org.jetlinks.core.server.session.DeviceSessionManager;
-import org.jetlinks.core.server.session.KeepOnlineSession;
+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;
@@ -37,9 +38,8 @@ import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.LongAdder;
-import java.util.function.Function;
 
-@Slf4j
+@Slf4j(topic = "system.tcp.gateway")
 class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGateway {
 
     @Getter
@@ -53,8 +53,6 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
 
     private final DeviceRegistry registry;
 
-    private final DecodedClientMessageHandler clientMessageHandler;
-
     private final DeviceSessionManager sessionManager;
 
     private final DeviceGatewayMonitor gatewayMonitor;
@@ -69,6 +67,8 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
 
     private Disposable disposable;
 
+    private final DeviceGatewayHelper helper;
+
     public TcpServerDeviceGateway(String id,
                                   String protocol,
                                   ProtocolSupports supports,
@@ -82,11 +82,10 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
         this.registry = deviceRegistry;
         this.supports = supports;
         this.tcpServer = tcpServer;
-        this.clientMessageHandler = clientMessageHandler;
         this.sessionManager = sessionManager;
+        this.helper = new DeviceGatewayHelper(registry, sessionManager, clientMessageHandler);
     }
 
-
     public Mono<ProtocolSupport> getProtocol() {
         return supports.getProtocol(protocol);
     }
@@ -153,96 +152,49 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
                 .filter(tcp -> started.get())
                 .publishOn(Schedulers.parallel())
                 .flatMap(this::handleTcpMessage)
-                .onErrorContinue((err, ignore) -> log.error(err.getMessage(), err))
+                .onErrorResume((err) -> {
+                    log.error(err.getMessage(), err);
+                    client.shutdown();
+                    return Mono.empty();
+                })
                 .then()
                 .doOnCancel(client::shutdown);
         }
 
         Mono<Void> handleTcpMessage(TcpMessage message) {
-
             return getProtocol()
                 .flatMap(pt -> pt.getMessageCodec(getTransport()))
-                .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(sessionRef.get(), message,registry)))
+                .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(sessionRef.get(), message, registry)))
                 .cast(DeviceMessage.class)
-                .doOnNext(msg-> gatewayMonitor.receivedMessage())
+                .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();
         }
 
         Mono<Void> handleDeviceMessage(DeviceMessage message) {
-            return registry
-                .getDevice(message.getDeviceId())
-                .switchIfEmpty(Mono.defer(() -> {
-                    if (processor.hasDownstreams()) {
-                        sink.next(message);
-                    }
-                    if (message instanceof DeviceRegisterMessage) {
-                        return clientMessageHandler
-                            .handleMessage(null, message)
-                            .then(Mono.empty());
-                    } else {
-                        log.warn("无法从tcp[{}]消息中获取设备信息:{}",address, message);
-                        return Mono.empty();
-                    }
-                }))
-                .flatMap(device -> {
-                    DeviceSession fSession = sessionManager.getSession(device.getDeviceId());
-                    //处理设备上线消息
-                    if (fSession == null) {
-                        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 (keepOnline) {
-                            fSession = new KeepOnlineSession(fSession, Duration.ofMillis(-1));
-                        } else {
-                            client.onDisconnect(() -> sessionManager.unregister(device.getDeviceId()));
-                        }
-                        sessionManager.register(fSession);
-                    }
-                    try {
-                        fSession.unwrap(TcpDeviceSession.class)
-                                .setClient(client);
-                    } catch (Throwable ignore) {
-
-                    }
-                    sessionRef.set(fSession);
-                    fSession.keepAlive();
-                    Duration timeout = message.getHeader(Headers.keepOnlineTimeoutSeconds).map(Duration::ofSeconds).orElse(keepaliveTimeout.get());
-                    if (timeout != null) {
-                        fSession.setKeepAliveTimeout(timeout);
-                    }
-                    fSession.keepAlive();
-                    if (message instanceof DeviceOnlineMessage) {
-                        return Mono.empty();
-                    }
-                    //设备下线
-                    if (message instanceof DeviceOfflineMessage) {
-                        sessionManager.unregister(device.getDeviceId());
-                        return Mono.empty();
-                    }
-                    message.addHeaderIfAbsent(Headers.clientAddress, String.valueOf(address));
-                    if (processor.hasDownstreams()) {
-                        sink.next(message);
-                    }
-                    return clientMessageHandler.handleMessage(device, message);
-                })
-                .then()
-                ;
+            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);
+                                         }),
+                                     () -> log.warn("无法从tcp[{}]消息中获取设备信息:{}", address, message)
+                )
+                .then();
         }
+
     }
 
+
     private void doStart() {
         if (started.getAndSet(true) || disposable != null) {
             return;
@@ -251,6 +203,7 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
             .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 -> {
@@ -261,7 +214,7 @@ class TcpServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGatew
 
     @Override
     public Flux<Message> onMessage() {
-        return processor.map(Function.identity());
+        return processor;
     }
 
     @Override

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

@@ -48,7 +48,7 @@ public class TcpServerDeviceGatewayProvider implements DeviceGatewayProvider {
 
     @Override
     public String getName() {
-        return "TCP服务设备网关";
+        return "TCP 透传接入";
     }
 
     @Override

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

@@ -12,16 +12,16 @@ import reactor.core.publisher.Mono;
 class UnknownTcpDeviceSession implements DeviceSession {
 
     @Getter
-    private String id;
+    private final String id;
 
-    private TcpClient client;
+    private final TcpClient client;
 
     @Getter
-    private Transport transport;
+    private final Transport transport;
 
     private long lastPingTime = System.currentTimeMillis();
 
-    private long connectTime = System.currentTimeMillis();
+    private final long connectTime = System.currentTimeMillis();
 
     UnknownTcpDeviceSession(String id, TcpClient client, Transport transport) {
         this.id = id;