From facee9342cbdd20f5af552e71104299619fe06f9 Mon Sep 17 00:00:00 2001
From: caoqianming <caoqianming@foxmail.com>
Date: Thu, 5 Jun 2025 11:15:16 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20cd=20BrokenPipeError=E6=97=B6=E5=B0=9D?=
 =?UTF-8?q?=E8=AF=95=E5=86=8D=E5=BB=BA=E7=AB=8B=E8=BF=9E=E6=8E=A52?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 apps/em/cd.py | 35 +++++++++++++++++++----------------
 1 file changed, 19 insertions(+), 16 deletions(-)

diff --git a/apps/em/cd.py b/apps/em/cd.py
index 02d3cf33..7c2a6050 100644
--- a/apps/em/cd.py
+++ b/apps/em/cd.py
@@ -8,6 +8,17 @@ from apps.utils.thread import MyThread
 def get_tyy_data_t(host, port):
     cd_thread_key = f"cd_thread_{host}_{port}"
     sc = None
+    def connect():
+        nonlocal sc
+        try:
+            sc = socket.socket()
+            sc.connect((host, int(port)))
+        except OSError as e:
+            cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"})
+        except ConnectionResetError:
+            cache.set(cd_thread_key, {"err_msg": "采集器重置了连接"})
+        except socket.timeout:
+            cache.set(cd_thread_key, {"err_msg": "采集器连接超时"})
     while True:
         cd_thread_val = cache.get(cd_thread_key, default=None)
         if cd_thread_val is None:
@@ -19,23 +30,15 @@ def get_tyy_data_t(host, port):
             break
         elif cd_thread_val == "get":
             if sc is None:
+                connect()
+            try:
+                sc.sendall(b"R")
+            except BrokenPipeError:
                 try:
-                    sc = socket.socket()
-                    sc.connect((host, int(port)))
-                except BrokenPipeError:
-                    try:
-                        sc.close()
-                    except Exception:
-                        pass
-                    sc = socket.socket()
-                    sc.connect((host, int(port)))
-                except OSError as e:
-                    cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"})
-                except ConnectionResetError:
-                    cache.set(cd_thread_key, {"err_msg": "采集器重置了连接"})
-                except socket.timeout:
-                    cache.set(cd_thread_key, {"err_msg": "采集器连接超时"})
-            sc.sendall(b"R")
+                    sc.close()
+                except Exception:
+                    pass
+                connect()
             resp = sc.recv(1024)
             if not resp:
                 cache.set(cd_thread_key, {"err_msg": f"采集器返回空数据-{str(resp)}"})