TCP网络组件

iot-components/iot-component-tcp/pom.xml

@@ -0,0 +1,106 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>iot-components</artifactId>
+        <groupId>cc.iotkit</groupId>
+        <version>0.4.2-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>iot-component-tcp</artifactId>
+
+    <properties>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+        <hsweb.expands.version>3.0.2</hsweb.expands.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.hswebframework</groupId>
+            <artifactId>hsweb-expands-script</artifactId>
+            <version>${hsweb.expands.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.luaj</groupId>
+            <artifactId>luaj-jse</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>iot-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>iot-component-base</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>iot-data-service</artifactId>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.2.4</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <artifactSet>
+                        <includes>
+                            <include>io.vertx:vertx-core</include>
+                            <include>org.luaj:luaj-jse</include>
+                        </includes>
+                    </artifactSet>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>11</source>
+                    <target>11</target>
+                    <forceJavacCompilerUse>true</forceJavacCompilerUse>
+                    <useIncrementalCompilation>false</useIncrementalCompilation>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/TcpDeviceComponent.java

@@ -0,0 +1,58 @@
+package cc.iotkit.comp;
+
+
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.comp.model.DeviceState;
+import cc.iotkit.comp.tcp.cilent.TcpClientDeviceComponent;
+import cc.iotkit.comp.tcp.server.TcpServerDeviceComponent;
+import cc.iotkit.converter.DeviceMessage;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Slf4j
+public class TcpDeviceComponent extends AbstractDeviceComponent {
+
+    private AbstractDeviceComponent tcpVerticle;
+
+    public void create(CompConfig config) {
+        Map maps = JsonUtil.parse(config.getOther(), Map.class);
+        String type = maps.get("type").toString();
+        if ("server".equals(type)) {
+            tcpVerticle = new TcpServerDeviceComponent();
+        } else {
+            tcpVerticle = new TcpClientDeviceComponent();
+        }
+        tcpVerticle.create(config);
+    }
+
+    @Override
+    public void start() {
+        tcpVerticle.start();
+    }
+
+    @Override
+    public void stop() {
+        tcpVerticle.stop();
+    }
+
+    @Override
+    public void destroy() {
+        tcpVerticle.destroy();
+    }
+
+    @Override
+    public void onDeviceStateChange(DeviceState state) {
+        tcpVerticle.onDeviceStateChange(state);
+    }
+
+
+    @Override
+    public DeviceMessage send(DeviceMessage message) {
+        return tcpVerticle.send(message);
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/cilent/TcpClientDeviceComponent.java

@@ -0,0 +1,75 @@
+package cc.iotkit.comp.tcp.cilent;
+
+import cc.iotkit.common.exception.BizException;
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.comp.AbstractDeviceComponent;
+import cc.iotkit.comp.CompConfig;
+import cc.iotkit.comp.model.DeviceState;
+import cc.iotkit.converter.DeviceMessage;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Slf4j
+public class TcpClientDeviceComponent extends AbstractDeviceComponent {
+
+    private Vertx vertx;
+    private TcpClientVerticle tcpClientVerticle;
+    private String deployedId;
+
+    public void create(CompConfig config) {
+        super.create(config);
+        vertx = Vertx.vertx();
+        TcpClinetConfig tcpClinetConfig = JsonUtil.parse(config.getOther(), TcpClinetConfig.class);
+        tcpClientVerticle = new TcpClientVerticle(tcpClinetConfig);
+    }
+
+    @Override
+    public void start() {
+        try {
+            tcpClientVerticle.setExecutor(getHandler());
+            Future<String> future = vertx.deployVerticle(tcpClientVerticle);
+            future.onSuccess((s -> {
+                        deployedId = s;
+                        log.info("tcp client start success, deployId:{}", s);
+                    }))
+                    .onFailure((e -> {
+                        log.error("tcp client start fail");
+                        e.printStackTrace();
+                    }));
+            future.succeeded();
+        } catch (Throwable e) {
+            throw new BizException("start client component error", e);
+        }
+
+    }
+
+    @Override
+    public void stop() {
+        tcpClientVerticle.stop();
+        Future<Void> future = vertx.undeploy(deployedId);
+        future.onSuccess(unused -> log.info("stop tcpserver component success"));
+    }
+
+    @Override
+    public void destroy() {
+    }
+
+    @Override
+    public void onDeviceStateChange(DeviceState state) {
+        if (DeviceState.STATE_OFFLINE.equals(state.getState())) {
+            tcpClientVerticle.offlineDevice(state.getDeviceName());
+        }
+    }
+
+
+    @Override
+    public DeviceMessage send(DeviceMessage message) {
+        tcpClientVerticle.sendMsg(message);
+        return message;
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/cilent/TcpClientVerticle.java

@@ -0,0 +1,131 @@
+package cc.iotkit.comp.tcp.cilent;
+
+import cc.iotkit.comp.IMessageHandler;
+import cc.iotkit.comp.tcp.parser.ParserStrategyBuilder;
+import cc.iotkit.converter.DeviceMessage;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.NetClient;
+import io.vertx.core.net.NetClientOptions;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author huangwenlong
+ * @version 1.0
+ * @date 2022/10/23 13:08
+ */
+@Slf4j
+public class TcpClientVerticle extends AbstractVerticle {
+
+    private TcpClinetConfig config;
+
+    private IMessageHandler executor;
+
+    private VertxTcpClient tcpClient;
+
+    private Map<String, ClientDevice> deviceMap = new ConcurrentHashMap();
+
+    public TcpClientVerticle(TcpClinetConfig config) {
+        this.config = config;
+    }
+
+    public void setExecutor(IMessageHandler executor) {
+        this.executor = executor;
+    }
+
+    @Override
+    public void start() {
+        tcpClient = new VertxTcpClient(UUID.randomUUID().toString(), false);
+        initConfig();
+        initClient();
+    }
+
+    @Override
+    public void stop() {
+        if (null != tcpClient) {
+            tcpClient.shutdown();
+        }
+    }
+
+    public void sendMsg(DeviceMessage msg) {
+        if (tcpClient != null) {
+            tcpClient.sendMessage(Buffer.buffer(msg.getContent().toString()));
+        }
+    }
+
+    public void offlineDevice(String deviceName) {
+        ClientDevice remove = deviceMap.remove(deviceName);
+    }
+
+    /**
+     * 创建配置文件
+     * 未链接成功就一直重连（每隔1分钟）
+     */
+    public void initConfig() {
+        if (config.getOptions() == null) {
+            NetClientOptions options = new NetClientOptions();
+            options.setReconnectAttempts(Integer.MAX_VALUE);
+            options.setReconnectInterval(60000L);
+            config.setOptions(options);
+        }
+        if (config.isSsl()) {
+            // 证书
+        }
+    }
+
+    private void initClient() {
+        NetClient netClient = vertx.createNetClient(config.getOptions());
+        tcpClient.setClient(netClient);
+        tcpClient.setKeepAliveTimeoutMs(Duration.ofMinutes(10).toMillis());
+        tcpClient.onDisconnect(() -> {
+            // 所有设备都离线
+            for (String deviceName : deviceMap.keySet()) {
+                // 发送离线消息
+                executor.onReceive(null, "disconnect", deviceName);
+            }
+        });
+        netClient.connect(config.getPort(), config.getHost(), result -> {
+            if (result.succeeded()) {
+                log.debug("connect tcp [{}:{}] success", config.getHost(), config.getPort());
+                tcpClient.setRecordParser(ParserStrategyBuilder.build(config.getParserType(), config.getParserConfiguration()));
+                tcpClient.setSocket(result.result());
+            } else {
+                log.error("connect tcp [{}:{}] error", config.getHost(), config.getPort(), result.cause());
+            }
+        });
+        // 设置收到消息处理
+        tcpClient.setReceiveHandler(buffer -> {
+            try {
+                executor.onReceive(null, "", buffer.toString(),
+                        result -> {
+                            if (!deviceMap.containsKey(result.getDeviceName())) {
+                                deviceMap.put(result.getDeviceName(), new ClientDevice(result.getDeviceName(), result.getProductKey()));
+                                // 有些设备并没有连接时报文，所以模拟一次 online
+                                HashMap<String, Object> map = new HashMap<>();
+                                map.put("deviceName", result.getDeviceName());
+                                executor.onReceive(map, "connect", buffer.toString());
+                            }
+                        });
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        });
+    }
+
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    class ClientDevice {
+        private String deviceName = "";
+        private String productKey = "";
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/cilent/TcpClinetConfig.java

@@ -0,0 +1,35 @@
+package cc.iotkit.comp.tcp.cilent;
+
+import io.vertx.core.net.NetClientOptions;
+import lombok.Data;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Data
+public class TcpClinetConfig {
+
+    private String id;
+
+    private NetClientOptions options;
+
+    private String host;
+
+    private int port;
+
+    private boolean ssl;
+
+    private String parserType;
+
+    // 解析参数
+    private Map<String, Object> parserConfiguration = new HashMap<>();
+
+    //服务实例数量(线程数)
+    private int instance = Runtime.getRuntime().availableProcessors();
+
+
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/cilent/VertxTcpClient.java

@@ -0,0 +1,181 @@
+package cc.iotkit.comp.tcp.cilent;
+
+import cc.iotkit.comp.IMessageHandler;
+import cc.iotkit.comp.tcp.parser.PayloadParser;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.NetClient;
+import io.vertx.core.net.NetSocket;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.codec.binary.Hex;
+import org.apache.commons.lang3.StringUtils;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Slf4j
+public class VertxTcpClient {
+    @Getter
+    private String id;
+    @Getter
+    private String deviceName = "";
+    @Setter
+    @Getter
+    private String parentName = "";
+    @Getter
+    private String productKey = "";
+    // 是否是服务端的连接客户端
+    private final boolean serverClient;
+    volatile PayloadParser payloadParser;
+    public volatile NetClient client;
+    public NetSocket socket;
+    private final List<Runnable> disconnectListener = new CopyOnWriteArrayList<>();
+    private IMessageHandler executor;
+    private Consumer<Buffer> receiveHandler;
+    @Setter
+    private long keepAliveTimeoutMs = Duration.ofMinutes(10).toMillis();
+    private volatile long lastKeepAliveTime = System.currentTimeMillis();
+
+    public VertxTcpClient(String id, boolean serverClient) {
+        this.id = id;
+        this.serverClient = serverClient;
+    }
+
+
+    public void keepAlive() {
+        lastKeepAliveTime = System.currentTimeMillis();
+    }
+
+    public boolean isOnline() {
+        return System.currentTimeMillis() - lastKeepAliveTime < keepAliveTimeoutMs;
+    }
+
+    public void setDeviceInfo(String deviceName, String productKey) {
+        this.deviceName = deviceName;
+        this.productKey = productKey;
+    }
+
+    public void setSocket(NetSocket socket) {
+        synchronized (this) {
+//            Objects.requireNonNull(payloadParser);
+            if (this.socket != null && this.socket != socket) {
+                this.socket.close();
+            }
+            this.socket = socket
+                    .closeHandler(v -> shutdown())
+                    .handler(buffer -> {
+                        if (log.isDebugEnabled()) {
+                            log.debug("handle tcp client[{}] payload:[{}]",
+                                    socket.remoteAddress(),
+                                    Hex.encodeHexString(buffer.getBytes()));
+                        }
+                        keepAlive();
+                        payloadParser.handle(buffer);
+                        if (this.socket != socket) {
+                            log.warn("tcp client [{}] memory leak ", socket.remoteAddress());
+                            socket.close();
+                        }
+                    });
+        }
+    }
+
+    /**
+     * 设置客户端消息解析器
+     *
+     * @param payloadParser 消息解析器
+     */
+    public void setRecordParser(PayloadParser payloadParser) {
+        synchronized (this) {
+            if (null != this.payloadParser && this.payloadParser != payloadParser) {
+                this.payloadParser.close();
+            }
+            this.payloadParser = payloadParser;
+            this.payloadParser
+                    .handlePayload()
+                    .onErrorContinue((err, res) -> {
+                        log.error(err.getMessage(), err);
+                        System.out.println(err.getMessage());
+                    })
+                    .subscribe(buffer -> {
+                        System.out.println(buffer.toString());
+                        receiveHandler.accept(buffer);
+                    });
+        }
+    }
+
+    public void onDisconnect(Runnable disconnected) {
+        disconnectListener.add(disconnected);
+    }
+
+
+    /**
+     * 设置消息处理器
+     */
+    public void setReceiveHandler(Consumer<Buffer> receiveHandler) {
+        this.receiveHandler = receiveHandler;
+    }
+
+    public void shutdown() {
+        log.debug("tcp client [{}] disconnect", getId());
+        synchronized (this) {
+            if (null != client) {
+                execute(client::close);
+                client = null;
+            }
+            if (null != socket) {
+                execute(socket::close);
+                this.socket = null;
+            }
+            // 粘包处理器
+            if (null != payloadParser) {
+                execute(payloadParser::close);
+                payloadParser = null;
+            }
+        }
+        for (Runnable runnable : disconnectListener) {
+            execute(runnable);
+        }
+        disconnectListener.clear();
+    }
+
+    public void sendMessage(Buffer buffer) {
+        socket.write(buffer, r -> {
+            keepAlive();
+            if (r.succeeded()) {
+                log.info("下行消息成功:{}", buffer.toString());
+            } else {
+                log.error("下行消息失败", r.cause());
+            }
+        });
+    }
+
+    private void execute(Runnable runnable) {
+        try {
+            runnable.run();
+        } catch (Exception e) {
+            log.warn("close tcp client error", e);
+        }
+    }
+
+    public void setClient(NetClient client) {
+        if (this.client != null && this.client != client) {
+            this.client.close();
+        }
+        keepAlive();
+        this.client = client;
+    }
+
+    /**
+     * 是否有父设备
+     */
+    public boolean hasParent() {
+        return StringUtils.isNotEmpty(parentName);
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/DelimitedPayloadParser.java

@@ -0,0 +1,55 @@
+package cc.iotkit.comp.tcp.parser;
+
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.parsetools.RecordParser;
+import org.apache.commons.lang3.StringEscapeUtils;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+
+import java.util.function.Function;
+
+/**
+ * 分隔符
+ *
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class DelimitedPayloadParser implements PayloadParser {
+
+    private String delimited;
+    private final EmitterProcessor<Buffer> processor = EmitterProcessor.create(false);
+    private final FluxSink<Buffer> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
+
+    private RecordParser recordParser;
+
+
+    public PayloadParser init(Object delimited) {
+        this.delimited = StringEscapeUtils.unescapeJava(delimited.toString());
+        this.reset();
+        return this;
+    }
+
+
+    @Override
+    public void handle(Buffer buffer) {
+        recordParser.handle(buffer);
+    }
+
+    @Override
+    public Flux<Buffer> handlePayload() {
+        return processor.map(Function.identity());
+    }
+
+    @Override
+    public void reset() {
+        this.recordParser = RecordParser.newDelimited(delimited);
+        // 塞入 skin pusher
+        this.recordParser.handler(sink::next);
+    }
+
+    @Override
+    public void close() {
+        processor.onComplete();
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/DirectPayloadParser.java

@@ -0,0 +1,41 @@
+package cc.iotkit.comp.tcp.parser;
+
+import io.vertx.core.buffer.Buffer;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+
+import java.util.function.Function;
+
+/**
+ * 固定长度
+ *
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class DirectPayloadParser implements PayloadParser {
+
+
+    EmitterProcessor<Buffer> processor = EmitterProcessor.create(false);
+
+
+    @Override
+    public PayloadParser init(Object param) {
+        return this;
+    }
+
+    @Override
+    public void handle(Buffer buffer) {
+        processor.onNext(buffer);
+    }
+
+    @Override
+    public Flux<Buffer> handlePayload() {
+        return processor.map(Function.identity());
+    }
+
+
+    @Override
+    public void close() {
+        processor.onComplete();
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/FixPayloadParser.java

@@ -0,0 +1,53 @@
+package cc.iotkit.comp.tcp.parser;
+
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.parsetools.RecordParser;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+
+import java.util.function.Function;
+
+/**
+ * 固定长度
+ *
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class FixPayloadParser implements PayloadParser {
+
+    private int size;
+    private final EmitterProcessor<Buffer> processor = EmitterProcessor.create(false);
+    private final FluxSink<Buffer> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
+
+    private RecordParser recordParser;
+
+
+    public PayloadParser init(Object size) {
+        this.size = (int) size;
+        this.reset();
+        return this;
+    }
+
+
+    @Override
+    public void handle(Buffer buffer) {
+        recordParser.handle(buffer);
+    }
+
+    @Override
+    public Flux<Buffer> handlePayload() {
+        return processor.map(Function.identity());
+    }
+
+    @Override
+    public void reset() {
+        this.recordParser = RecordParser.newFixed(size);
+        this.recordParser.handler(sink::next);
+    }
+
+    @Override
+    public void close() {
+        processor.onComplete();
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/ParserStrategyBuilder.java

@@ -0,0 +1,35 @@
+package cc.iotkit.comp.tcp.parser;
+
+import cc.iotkit.comp.tcp.parser.builder.DelimitedPayloadBuilder;
+import cc.iotkit.comp.tcp.parser.builder.DirectPayloadBuilder;
+import cc.iotkit.comp.tcp.parser.builder.FixPayloadBuilder;
+import cc.iotkit.comp.tcp.parser.builder.ScriptPayloadBuilder;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class ParserStrategyBuilder {
+    private static Map<String, PayloadParserBuilderStrategy> strategyMap = new ConcurrentHashMap<>();
+
+    static  {
+        register(new DelimitedPayloadBuilder());
+        register(new DirectPayloadBuilder());
+        register(new FixPayloadBuilder());
+        register(new ScriptPayloadBuilder());
+    }
+
+    public static PayloadParser build(String type, Map<String, Object> configuration) {
+        return Optional.ofNullable(strategyMap.get(type))
+                .map(builder -> builder.build(configuration))
+                .orElseThrow(() -> new UnsupportedOperationException("unsupported parser:" + type));
+    }
+
+    private static void register(PayloadParserBuilderStrategy strategy) {
+        strategyMap.put(strategy.getType().getText(), strategy);
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/PayloadParser.java

@@ -0,0 +1,32 @@
+package cc.iotkit.comp.tcp.parser;
+
+import io.vertx.core.buffer.Buffer;
+import reactor.core.publisher.Flux;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public interface PayloadParser {
+
+
+    PayloadParser init(Object param);
+
+    void handle(Buffer buffer);
+
+    /**
+     * 订阅完整的数据包流,每一个元素为一个完整的数据包
+     *
+     * @return 完整数据包流
+     */
+    Flux<Buffer> handlePayload();
+
+
+    /**
+     * 重置规则
+     */
+    default void reset() {
+    }
+
+    void close();
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/PayloadParserBuilderStrategy.java

@@ -0,0 +1,17 @@
+package cc.iotkit.comp.tcp.parser;
+
+
+
+import cc.iotkit.comp.tcp.parser.enums.PayloadParserType;
+
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public interface PayloadParserBuilderStrategy {
+    PayloadParserType getType();
+
+    PayloadParser build(Map<String, Object> parserConfiguration);
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/ScriptPayloadParser.java

@@ -0,0 +1,41 @@
+package cc.iotkit.comp.tcp.parser;
+
+import io.vertx.core.buffer.Buffer;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+
+import java.util.function.Function;
+
+/**
+ * 固定长度
+ *
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class ScriptPayloadParser implements PayloadParser {
+
+
+    EmitterProcessor<Buffer> processor = EmitterProcessor.create(false);
+
+
+    @Override
+    public PayloadParser init(Object param) {
+        return this;
+    }
+
+    @Override
+    public void handle(Buffer buffer) {
+        processor.onNext(buffer);
+    }
+
+    @Override
+    public Flux<Buffer> handlePayload() {
+        return processor.map(Function.identity());
+    }
+
+
+    @Override
+    public void close() {
+        processor.onComplete();
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/builder/DelimitedPayloadBuilder.java

@@ -0,0 +1,24 @@
+package cc.iotkit.comp.tcp.parser.builder;
+
+import cc.iotkit.comp.tcp.parser.DelimitedPayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParserBuilderStrategy;
+import cc.iotkit.comp.tcp.parser.enums.PayloadParserType;
+
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class DelimitedPayloadBuilder implements PayloadParserBuilderStrategy {
+    @Override
+    public PayloadParserType getType() {
+        return PayloadParserType.DELIMITED;
+    }
+
+    @Override
+    public PayloadParser build(Map<String, Object> parserConfiguration) {
+        return new DelimitedPayloadParser().init(parserConfiguration.get("delimited"));
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/builder/DirectPayloadBuilder.java

@@ -0,0 +1,24 @@
+package cc.iotkit.comp.tcp.parser.builder;
+
+import cc.iotkit.comp.tcp.parser.DirectPayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParserBuilderStrategy;
+import cc.iotkit.comp.tcp.parser.enums.PayloadParserType;
+
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class DirectPayloadBuilder implements PayloadParserBuilderStrategy {
+    @Override
+    public PayloadParserType getType() {
+        return PayloadParserType.DIRECT;
+    }
+
+    @Override
+    public PayloadParser build(Map<String, Object> parserConfiguration) {
+        return new DirectPayloadParser().init(null);
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/builder/FixPayloadBuilder.java

@@ -0,0 +1,24 @@
+package cc.iotkit.comp.tcp.parser.builder;
+
+import cc.iotkit.comp.tcp.parser.FixPayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParserBuilderStrategy;
+import cc.iotkit.comp.tcp.parser.enums.PayloadParserType;
+
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class FixPayloadBuilder implements PayloadParserBuilderStrategy {
+    @Override
+    public PayloadParserType getType() {
+        return PayloadParserType.FIXED_LENGTH;
+    }
+
+    @Override
+    public PayloadParser build(Map<String, Object> parserConfiguration) {
+        return new FixPayloadParser().init(parserConfiguration.get("fix"));
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/builder/ScriptPayloadBuilder.java

@@ -0,0 +1,24 @@
+package cc.iotkit.comp.tcp.parser.builder;
+
+import cc.iotkit.comp.tcp.parser.PayloadParser;
+import cc.iotkit.comp.tcp.parser.PayloadParserBuilderStrategy;
+import cc.iotkit.comp.tcp.parser.ScriptPayloadParser;
+import cc.iotkit.comp.tcp.parser.enums.PayloadParserType;
+
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+public class ScriptPayloadBuilder implements PayloadParserBuilderStrategy {
+    @Override
+    public PayloadParserType getType() {
+        return PayloadParserType.SCRIPT;
+    }
+
+    @Override
+    public PayloadParser build(Map<String, Object> parserConfiguration) {
+        return new ScriptPayloadParser().init(parserConfiguration.get("script"));
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/parser/enums/PayloadParserType.java

@@ -0,0 +1,23 @@
+package cc.iotkit.comp.tcp.parser.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Getter
+@AllArgsConstructor
+public enum PayloadParserType {
+    DIRECT("不处理"),
+
+    FIXED_LENGTH("固定长度"),
+
+    DELIMITED("分隔符"),
+
+    SCRIPT("自定义脚本")
+    ;
+
+    private String text;
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/server/TcpServerConfig.java

@@ -0,0 +1,43 @@
+package cc.iotkit.comp.tcp.server;
+
+import io.vertx.core.net.NetServerOptions;
+import io.vertx.core.net.SocketAddress;
+import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Data
+public class TcpServerConfig {
+
+    private String id;
+
+    private NetServerOptions options;
+
+    private String host;
+
+    private int port;
+
+    private boolean ssl;
+
+    private String parserType;
+
+    // 解析参数
+    private Map<String, Object> parserConfiguration = new HashMap<>();
+
+    //服务实例数量(线程数)
+    private int instance = Runtime.getRuntime().availableProcessors();
+
+    public SocketAddress createSocketAddress() {
+        if (StringUtils.isEmpty(host)) {
+            host = "localhost";
+        }
+        return SocketAddress.inetSocketAddress(port, host);
+    }
+
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/server/TcpServerDeviceComponent.java

@@ -0,0 +1,78 @@
+package cc.iotkit.comp.tcp.server;
+
+import cc.iotkit.common.exception.BizException;
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.comp.AbstractDeviceComponent;
+import cc.iotkit.comp.CompConfig;
+import cc.iotkit.comp.model.DeviceState;
+import cc.iotkit.converter.DeviceMessage;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Slf4j
+public class TcpServerDeviceComponent extends AbstractDeviceComponent {
+
+    private Vertx vertx;
+    private TcpServerVerticle tcpServerVerticle;
+    private String deployedId;
+
+    public void create(CompConfig config) {
+        super.create(config);
+        vertx = Vertx.vertx();
+        TcpServerConfig serverConfig = JsonUtil.parse(config.getOther(), TcpServerConfig.class);
+        tcpServerVerticle = new TcpServerVerticle(serverConfig);
+    }
+
+    @Override
+    public void start() {
+        try {
+            tcpServerVerticle.setExecutor(getHandler());
+            Future<String> future = vertx.deployVerticle(tcpServerVerticle);
+            future.onSuccess((s -> {
+                        deployedId = s;
+                        log.info("tcp server start success, deployId:{}", s);
+                    }))
+                    .onFailure((e -> {
+                        log.error("tcp server start fail");
+                        e.printStackTrace();
+                    }));
+            future.succeeded();
+        }catch (Throwable e){
+            throw new BizException("start tcpserver component error", e);
+        }
+
+    }
+
+    @Override
+    public void stop() {
+        tcpServerVerticle.stop();
+        Future<Void> future = vertx.undeploy(deployedId);
+        future.onSuccess(unused -> log.info("stop tcpserver component success"));
+    }
+
+    @Override
+    public void destroy() {
+    }
+
+    @Override
+    public void onDeviceStateChange(DeviceState state) {
+        if (DeviceState.STATE_OFFLINE.equals(state.getState())){
+            tcpServerVerticle.offlineDevice(state.getDeviceName());
+        }else if(DeviceState.STATE_ONLINE.equals(state.getState())){
+            tcpServerVerticle.onlineDevice(state.getDeviceName(),
+                    state.getParent() != null ? state.getParent().getDeviceName() : null);
+        }
+    }
+
+
+    @Override
+    public DeviceMessage send(DeviceMessage message) {
+        tcpServerVerticle.sendMsg(message);
+        return message;
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/server/TcpServerVerticle.java

@@ -0,0 +1,255 @@
+package cc.iotkit.comp.tcp.server;
+
+
+import cc.iotkit.comp.IMessageHandler;
+import cc.iotkit.comp.tcp.cilent.VertxTcpClient;
+import cc.iotkit.comp.tcp.parser.ParserStrategyBuilder;
+import cc.iotkit.comp.tcp.parser.PayloadParser;
+import cc.iotkit.converter.DeviceMessage;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.NetServer;
+import io.vertx.core.net.NetServerOptions;
+import io.vertx.core.net.NetSocket;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * @author huangwenl
+ * @date 2022-10-13
+ */
+@Slf4j
+public class TcpServerVerticle extends AbstractVerticle {
+
+    @Getter
+    private TcpServerConfig config;
+
+    private IMessageHandler executor;
+
+    private VertxTcpServer tcpServer;
+
+    private String id;
+
+    private Map<String, VertxTcpClient> clientMap = new ConcurrentHashMap();
+
+    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
+
+    @Setter
+    private long keepAliveTimeout = Duration.ofMinutes(10).toMillis();
+
+    private Collection<NetServer> tcpServers;
+
+    public TcpServerVerticle(TcpServerConfig config) {
+        this.config = config;
+    }
+
+    public void setExecutor(IMessageHandler executor) {
+        this.executor = executor;
+    }
+
+    @Override
+    public void start() {
+        tcpServer = new VertxTcpServer();
+        initConfig();
+        initTcpServer();
+        keepClientTask();
+    }
+
+    @Override
+    public void stop() {
+        tcpServer.shutdown();
+        scheduledThreadPoolExecutor.shutdown();
+    }
+
+    /**
+     * 创建配置文件
+     */
+    public void initConfig() {
+        if (config.getOptions() == null) {
+            config.setOptions(new NetServerOptions());
+        }
+        if (config.isSsl()) {
+            // 证书
+        }
+    }
+
+
+    private void initTcpServer() {
+        int instance = Math.max(2, config.getInstance());
+        List<NetServer> instances = new ArrayList<>(instance);
+        for (int i = 0; i < instance; i++) {
+            instances.add(vertx.createNetServer(config.getOptions()));
+        }
+        // 根据解析类型配置数据解析器
+        tcpServer.setParserSupplier(() -> ParserStrategyBuilder.build(config.getParserType(), config.getParserConfiguration()));
+        tcpServer.setServer(instances);
+        // 针对JVM做的多路复用优化
+        // 多个server listen同一个端口，每个client连接的时候vertx会分配
+        // 一个connection只能在一个server中处理
+        for (NetServer netServer : instances) {
+            netServer.listen(config.createSocketAddress(), result -> {
+                if (result.succeeded()) {
+                    log.info("tcp server startup on {}", result.result().actualPort());
+                } else {
+                    log.error("startup tcp server error", result.cause());
+                }
+            });
+        }
+    }
+
+    public void offlineDevice(String deviceName) {
+        VertxTcpClient client = clientMap.get(deviceName);
+        if (client != null) {
+            client.shutdown();
+        }
+    }
+
+    public void onlineDevice(String deviceName, String parentName) {
+        VertxTcpClient client = clientMap.get(deviceName);
+        if (client != null) {
+            client.setParentName(parentName);
+        }
+    }
+
+    public void sendMsg(DeviceMessage msg) {
+        VertxTcpClient tcpClient = clientMap.get(msg.getDeviceName());
+        if (tcpClient != null) {
+            tcpClient.sendMessage(Buffer.buffer(msg.getContent().toString()));
+        }
+    }
+
+    public void keepClientTask() {
+        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(2);
+
+        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
+            log.info("保活任务开始！");
+            Set<String> clients = new HashSet(clientMap.keySet());
+            for (String key : clients) {
+                VertxTcpClient client = clientMap.get(key);
+                if (!client.isOnline()) {
+                    client.shutdown();
+                }
+            }
+        }, 1000, keepAliveTimeout, TimeUnit.MILLISECONDS);
+    }
+
+
+    class VertxTcpServer {
+
+        private Supplier<PayloadParser> parserSupplier;
+
+        /**
+         * 为每个NetServer添加connectHandler
+         *
+         * @param servers 创建的所有NetServer
+         */
+        public void setServer(Collection<NetServer> servers) {
+            if (tcpServers != null && !tcpServers.isEmpty()) {
+                shutdown();
+            }
+            tcpServers = servers;
+            for (NetServer tcpServer : tcpServers) {
+                tcpServer.connectHandler(this::acceptTcpConnection);
+            }
+        }
+
+        /**
+         * TCP连接处理逻辑
+         *
+         * @param socket socket
+         */
+        protected void acceptTcpConnection(NetSocket socket) {
+            // 客户端连接处理
+            String clientId = id + "_" + socket.remoteAddress();
+            VertxTcpClient client = new VertxTcpClient(clientId, true);
+            client.setKeepAliveTimeoutMs(keepAliveTimeout);
+            try {
+                // TCP异常和关闭处理
+                socket.exceptionHandler(err -> {
+                    log.error("tcp server client [{}] error", socket.remoteAddress(), err);
+                }).closeHandler((nil) -> {
+                    log.debug("tcp server client [{}] closed", socket.remoteAddress());
+                    client.shutdown();
+                });
+                // 这个地方是在TCP服务初始化的时候设置的 parserSupplier
+                client.setKeepAliveTimeoutMs(keepAliveTimeout);
+                client.setRecordParser(parserSupplier.get());
+                client.setSocket(socket);
+                client.onDisconnect(() -> {
+                    clientDisconnect(client.getDeviceName());
+                });
+                // 设置收到消息处理
+                client.setReceiveHandler(buffer -> {
+                    System.out.println(buffer.toString());
+                    try {
+                        executor.onReceive(null, "", buffer.toString(),
+                                result -> {
+                                    if (result != null && !clientMap.containsKey(result.getDeviceName())) {
+                                        client.setDeviceInfo(result.getDeviceName(), result.getProductKey());
+                                        clientMap.put(result.getDeviceName(), client);
+                                        // 有些设备并没有连接时报文，所以模拟一次 online
+                                        HashMap<String, Object> map = new HashMap<>();
+                                        map.put("deviceName", result.getDeviceName());
+                                        executor.onReceive(map, "connect", buffer.toString());
+                                    }
+                                });
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+
+                });
+                log.debug("accept tcp client [{}] connection", socket.remoteAddress());
+            } catch (Exception e) {
+                log.error("create tcp server client error", e);
+                client.shutdown();
+            }
+        }
+
+        public void setParserSupplier(Supplier<PayloadParser> parserSupplier) {
+            this.parserSupplier = parserSupplier;
+        }
+
+        public void shutdown() {
+            if (null != tcpServers) {
+                for (NetServer tcpServer : tcpServers) {
+                    execute(tcpServer::close);
+                }
+                tcpServers = null;
+            }
+        }
+
+        private void execute(Runnable runnable) {
+            try {
+                runnable.run();
+            } catch (Exception e) {
+                log.warn("close tcp server error", e);
+            }
+        }
+
+        /**
+         * 递归断开连接
+         */
+        private void clientDisconnect(String deviceName) {
+            VertxTcpClient remove = clientMap.remove(deviceName);
+            if (null != remove) {
+                // 发送离线消息
+                executor.onReceive(null, "disconnect", deviceName);
+                // 移除子设备
+                if (remove.hasParent()) {
+                    List<VertxTcpClient> childClients = clientMap.values().stream().filter(cl -> cl.hasParent() && cl.getParentName()
+                            .equals(remove.getParentName())).collect(Collectors.toList());
+                    childClients.forEach(child -> clientMap.remove(child.getDeviceName()));
+                }
+            }
+        }
+    }
+}

iot-components/iot-component-tcp/src/main/java/cc/iotkit/comp/tcp/server/VertxTcpServer.java

@@ -0,0 +1,160 @@
+package cc.iotkit.comp.tcp.server;//package cc.iotkit.comp.tcp.server;
+//
+//import cc.iotkit.comp.IMessageHandler;
+//import cc.iotkit.comp.tcp.cilent.VertxTcpClient;
+//import cc.iotkit.comp.tcp.parser.PayloadParser;
+//import cc.iotkit.converter.DeviceMessage;
+//import io.vertx.core.buffer.Buffer;
+//import io.vertx.core.net.NetServer;
+//import io.vertx.core.net.NetSocket;
+//import lombok.Setter;
+//import lombok.extern.slf4j.Slf4j;
+//
+//import java.time.Duration;
+//import java.util.Collection;
+//import java.util.HashMap;
+//import java.util.List;
+//import java.util.Map;
+//import java.util.concurrent.ConcurrentHashMap;
+//import java.util.function.Supplier;
+//import java.util.stream.Collectors;
+//
+///**
+// * @author huangwenl
+// * @date 2022-10-13
+// */
+//@Slf4j
+//public class VertxTcpServer {
+//
+//    private String id;
+//
+//    private Supplier<PayloadParser> parserSupplier;
+//
+//    private Map<String, VertxTcpClient> clientMap = new ConcurrentHashMap();
+//    private IMessageHandler executor;
+//    private Collection<NetServer> tcpServers;
+//    @Setter
+//    private long keepAliveTimeout = Duration.ofMinutes(10).toMillis();
+//
+//    public VertxTcpServer(String id) {
+//        this.id = id;
+//    }
+//
+//    /**
+//     * 为每个NetServer添加connectHandler
+//     *
+//     * @param servers 创建的所有NetServer
+//     */
+//    public void setServer(Collection<NetServer> servers) {
+//        if (this.tcpServers != null && !this.tcpServers.isEmpty()) {
+//            shutdown();
+//        }
+//        this.tcpServers = servers;
+//
+//        for (NetServer tcpServer : this.tcpServers) {
+//            tcpServer.connectHandler(this::acceptTcpConnection);
+//        }
+//    }
+//
+//    /**
+//     * TCP连接处理逻辑
+//     *
+//     * @param socket socket
+//     */
+//    protected void acceptTcpConnection(NetSocket socket) {
+//        // 客户端连接处理
+//        String clientId = id + "_" + socket.remoteAddress();
+//        VertxTcpClient client = new VertxTcpClient(clientId, true);
+//        client.setKeepAliveTimeoutMs(keepAliveTimeout);
+//        try {
+//            // TCP异常和关闭处理
+//            socket.exceptionHandler(err -> {
+//                log.error("tcp server client [{}] error", socket.remoteAddress(), err);
+//            }).closeHandler((nil) -> {
+//                log.debug("tcp server client [{}] closed", socket.remoteAddress());
+//                client.shutdown();
+//            });
+//            // 这个地方是在TCP服务初始化的时候设置的 parserSupplier
+//            client.setRecordParser(parserSupplier.get());
+//            client.setSocket(socket);
+//            client.onDisconnect(() -> {
+//                clientDisconnect(client.getDeviceName());
+//            });
+//            // 设置收到消息处理
+//            client.setReceiveHandler(buffer -> {
+//                executor.onReceive(null, "", buffer.toString(),
+//                        result -> {
+//                            if (!clientMap.containsKey(result.getDeviceName())) {
+//                                client.setDeviceInfo(result.getDeviceName(), result.getData().getParent().getDeviceName(),
+//                                        result.getProductKey());
+//                                clientMap.put(result.getDeviceName(), client);
+//                                // 有些设备并没有连接时报文，所以模拟一次 online
+//                                HashMap<String, Object> map = new HashMap<>();
+//                                map.put("deviceName", result.getDeviceName());
+//                                executor.onReceive(map, "online", "");
+//                            }
+//                        });
+//            });
+////            clientMap.put(clientId, client);
+//            log.debug("accept tcp client [{}] connection", socket.remoteAddress());
+//        } catch (Exception e) {
+//            log.error("create tcp server client error", e);
+//            client.shutdown();
+//        }
+//    }
+//
+//    public void setParserSupplier(Supplier<PayloadParser> parserSupplier) {
+//        this.parserSupplier = parserSupplier;
+//    }
+//
+//    public void shutdown() {
+//        if (null != tcpServers) {
+//            for (NetServer tcpServer : tcpServers) {
+//                execute(tcpServer::close);
+//            }
+//            tcpServers = null;
+//        }
+//    }
+//
+//    private void execute(Runnable runnable) {
+//        try {
+//            runnable.run();
+//        } catch (Exception e) {
+//            log.warn("close tcp server error", e);
+//        }
+//    }
+//
+//
+//    public void sendMsg(DeviceMessage message) {
+//        VertxTcpClient client = clientMap.get(message.getDeviceName());
+//        if (client != null) {
+//            client.sendMessage(Buffer.buffer(message.getContent().toString()));
+//        }
+//    }
+//
+//    /**
+//     * 递归断开连接
+//     */
+//    private void clientDisconnect(String deviceName) {
+//        VertxTcpClient remove = clientMap.remove(deviceName);
+//        if (null != remove) {
+//            executor.onReceive(null, "disconnect", deviceName);
+//            if (remove.hasParent()) {
+//                List<VertxTcpClient> childClients = clientMap.values().stream().filter(cl -> cl.hasParent() && cl.getParentName()
+//                        .equals(remove.getParentName())).collect(Collectors.toList());
+//                childClients.forEach(child -> clientDisconnect(child.getDeviceName()));
+//            }
+//        }
+//    }
+//
+//    public void offlineDev(String deviceName) {
+//        VertxTcpClient remove = clientMap.remove(deviceName);
+//        if (null != remove) {
+//            if (remove.hasParent()) {
+//                List<VertxTcpClient> childClients = clientMap.values().stream().filter(cl -> cl.hasParent() && cl.getParentName()
+//                        .equals(remove.getParentName())).collect(Collectors.toList());
+//                childClients.forEach(child -> offlineDev(child.getDeviceName()));
+//            }
+//        }
+//    }
+//}

iot-components/iot-component-tcp/src/main/resources/component.spi

@@ -0,0 +1 @@
+cc.iotkit.comp.TcpDeviceComponent

iot-components/pom.xml

@@ -18,6 +18,7 @@
         <module>iot-emqx-component</module>
         <module>iot-component-base</module>
         <module>iot-http-biz-component</module>
+        <module>iot-component-tcp</module>
 <!--        <module>iot-ctwing-component</module>-->
     </modules>