From 945bf2ea82a1e33aa6c338102faaa9836473343b Mon Sep 17 00:00:00 2001
From: zhouhao <zh.sqy@qq.com>
Date: Mon, 21 Oct 2024 17:25:56 +0800
Subject: [PATCH] =?UTF-8?q?refactor(=E5=9F=BA=E7=A1=80=E6=A8=A1=E5=9D=97):?=
 =?UTF-8?q?=20=E4=BC=98=E5=8C=96=E8=AE=BE=E5=A4=87=E8=BF=9E=E6=8E=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../network/http/device/HttpDeviceSession.java       | 12 +++++++++---
 .../community/network/tcp/client/VertxTcpClient.java |  2 +-
 2 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java
index 84d86f3e..3cab4988 100755
--- a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java
+++ b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java
@@ -73,8 +73,11 @@ class HttpDeviceSession implements DeviceSession {
     @Override
     public Mono<Boolean> send(EncodedMessage encodedMessage) {
         //未建立websocket链接,不支持此类消息.
-        if(websocket == null){
-            return Mono.error(new DeviceOperationException(ErrorCode.UNSUPPORTED_MESSAGE));
+        if (websocket == null) {
+            return Mono.error(new DeviceOperationException.NoStackTrace(ErrorCode.UNSUPPORTED_MESSAGE));
+        }
+        if (!websocket.isAlive()) {
+            return Mono.error(new DeviceOperationException.NoStackTrace(ErrorCode.CONNECTION_LOST));
         }
         if (encodedMessage instanceof WebSocketMessage) {
             return websocket
@@ -99,7 +102,10 @@ class HttpDeviceSession implements DeviceSession {
 
     @Override
     public void close() {
-
+        //断开websocket连接
+        if (websocket != null) {
+            websocket.close().subscribe();
+        }
     }
 
     @Override
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 a1a6faf1..e92433f2 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
@@ -46,7 +46,7 @@ public class VertxTcpClient implements TcpClient {
 
     private final List<Runnable> disconnectListener = new CopyOnWriteArrayList<>();
 
-    private final Sinks.Many<TcpMessage> sink = Reactors.createMany();
+    private final Sinks.Many<TcpMessage> sink = Reactors.createMany(Integer.MAX_VALUE, false);
 
     private final boolean serverClient;