class TcpConnection implements DeviceGatewayContext {
        final TcpClient client;
        final AtomicReference<DeviceSession> sessionRef = new AtomicReference<>();
        final InetSocketAddress address;
        Disposable legalityChecker;

        TcpConnection(TcpClient client) {
            this.client = client;
            this.address = client.getRemoteAddress();
            monitor.totalConnection(counter.sum());
            client.onDisconnect(() -> {
                counter.decrement();
                monitor.disconnected();
                monitor.totalConnection(counter.sum());
                //check session
                DeviceSession session = sessionRef.get();
                if (session.getDeviceId() != null) {
                    sessionManager
                        .getSession(session.getDeviceId())
                        .subscribe();
                }

            });
            monitor.connected();

            sessionRef.set(new UnknownTcpDeviceSession(client.getId(), client, DefaultTransport.TCP, monitor));

            legalityChecker = Schedulers
                .parallel()
                .schedule(this::checkLegality, connectCheckTimeout.toMillis(), TimeUnit.MILLISECONDS);
        }

        public void checkLegality() {<FILL_FUNCTION_BODY>}

        Mono<Void> accept() {
            return getProtocol()
                .flatMap(protocol -> protocol.onClientConnect(DefaultTransport.TCP, client, this))
                .then(
                    client
                        .subscribe()
                        .filter(tcp -> started.get())
                        .concatMap(this::handleTcpMessage)
                        .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(DefaultTransport.TCP))
                .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(
                    sessionRef.get(), message, registry, msg -> handleDeviceMessage(msg).then())))
                .cast(DeviceMessage.class)
                .concatMap(msg -> this
                    .handleDeviceMessage(msg)
                    .as(MonoTracer.create(
                        DeviceTracer.SpanName.decode(msg.getDeviceId()),
                        (span, _msg) -> span.setAttributeLazy(DeviceTracer.SpanKey.message, _msg::toString))))
                .onErrorResume((err) -> {
                    log.error("Handle TCP[{}] message failed:\n{}",
                              address,
                              message
                        , err);
                    return Mono.fromRunnable(client::reset);
                })
                .subscribeOn(Schedulers.parallel())
                .then();
        }

        Mono<DeviceMessage> handleDeviceMessage(DeviceMessage message) {
            Disposable checker = legalityChecker;
            if (checker != null) {
                checker.dispose();
                legalityChecker = null;
            }
            monitor.receivedMessage();
            return helper
                .handleDeviceMessage(
                    message,
                    device -> new TcpDeviceSession(device, client,DefaultTransport.TCP, monitor),
                    session -> {
                        TcpDeviceSession deviceSession = session.unwrap(TcpDeviceSession.class);
                        deviceSession.setClient(client);
                        sessionRef.set(deviceSession);
                    },
                    () -> log.warn("TCP{}: The device[{}] in the message body does not exist:{}", address, message.getDeviceId(), message)
                )
                .thenReturn(message);
        }

        @Override
        public Mono<DeviceOperator> getDevice(String deviceId) {
            return registry.getDevice(deviceId);
        }

        @Override
        public Mono<DeviceProductOperator> getProduct(String productId) {
            return registry.getProduct(productId);
        }

        @Override
        public Mono<Void> onMessage(DeviceMessage message) {
            return handleDeviceMessage(message).then();
        }
    }

            //超过时间还未获取到任何设备则认为连接不合法，自动断开连接
            if ((sessionRef.get() instanceof UnknownTcpDeviceSession)) {
                log.info("tcp [{}] connection is illegal, close it.", address);
                try {
                    client.disconnect();
                } catch (Throwable ignore) {
                }
            }
        
<methods>public void <init>(java.lang.String) ,public final void doOnShutdown(reactor.core.Disposable) ,public final void doOnStateChange(BiConsumer<org.jetlinks.community.gateway.GatewayState,org.jetlinks.community.gateway.GatewayState>) ,public final java.lang.String getId() ,public final org.jetlinks.community.gateway.GatewayState getState() ,public final boolean isAlive() ,public Flux<org.jetlinks.core.message.Message> onMessage() ,public final Mono<java.lang.Void> pause() ,public final Mono<java.lang.Void> shutdown() ,public final synchronized Mono<java.lang.Void> startup() <variables>private static final AtomicReferenceFieldUpdater<org.jetlinks.community.gateway.AbstractDeviceGateway,org.jetlinks.community.gateway.GatewayState> STATE,private final non-sealed java.lang.String id,protected final non-sealed org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor monitor,private volatile org.jetlinks.community.gateway.GatewayState state,private final List<BiConsumer<org.jetlinks.community.gateway.GatewayState,org.jetlinks.community.gateway.GatewayState>> stateListener