From f6dc39a6683f7604ed89dc25f724933d6bc35ded Mon Sep 17 00:00:00 2001
From: zhou-hao <admin@hsweb.me>
Date: Mon, 30 Nov 2020 19:32:57 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96tcp=20client=20=E5=85=B3?=
 =?UTF-8?q?=E9=97=AD=E6=97=B6,=E9=87=8A=E6=94=BEFluxProcessor?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../community/network/tcp/client/VertxTcpClient.java      | 8 +++++++-
 .../network/tcp/client/VertxTcpClientProvider.java        | 2 +-
 .../community/network/tcp/server/VertxTcpServer.java      | 2 +-
 3 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClient.java b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClient.java
index 6bfef4a2..bae531dc 100644
--- a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClient.java
+++ b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClient.java
@@ -48,6 +48,8 @@ public class VertxTcpClient implements TcpClient {
 
     private final FluxSink<TcpMessage> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
 
+    private final boolean serverClient;
+
     @Override
     public void keepAlive() {
         lastKeepAliveTime = System.currentTimeMillis();
@@ -75,8 +77,9 @@ public class VertxTcpClient implements TcpClient {
         return true;
     }
 
-    public VertxTcpClient(String id) {
+    public VertxTcpClient(String id,boolean serverClient) {
         this.id = id;
+        this.serverClient=serverClient;
     }
 
     protected void received(TcpMessage message) {
@@ -136,6 +139,9 @@ public class VertxTcpClient implements TcpClient {
             execute(runnable);
         }
         disconnectListener.clear();
+        if(serverClient){
+            processor.onComplete();
+        }
     }
 
     public void setClient(NetClient client) {
diff --git a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClientProvider.java b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClientProvider.java
index 03a0e60f..3511a73e 100644
--- a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClientProvider.java
+++ b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClientProvider.java
@@ -43,7 +43,7 @@ public class VertxTcpClientProvider implements NetworkProvider<TcpClientProperti
     @Nonnull
     @Override
     public VertxTcpClient createNetwork(@Nonnull TcpClientProperties properties) {
-        VertxTcpClient client = new VertxTcpClient(properties.getId());
+        VertxTcpClient client = new VertxTcpClient(properties.getId(),false);
 
         initClient(client, properties);
 
diff --git a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/server/VertxTcpServer.java b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/server/VertxTcpServer.java
index e18ba030..287c7e92 100644
--- a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/server/VertxTcpServer.java
+++ b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/server/VertxTcpServer.java
@@ -81,7 +81,7 @@ public class VertxTcpServer implements TcpServer {
             socket.close();
             return;
         }
-        VertxTcpClient client = new VertxTcpClient(id + "_" + socket.remoteAddress());
+        VertxTcpClient client = new VertxTcpClient(id + "_" + socket.remoteAddress(), true);
         client.setKeepAliveTimeoutMs(keepAliveTimeout);
         try {
             socket.exceptionHandler(err -> {