lianlonggangandClaude
0b13de14eb
- 配置化：gateway.url/socket.path/download.prefix（网关基础地址统一推导 ws 与下载地址）、 main.home 主目录（jar/pid 相对它推导）、order.basic；properties 中文 \uXXXX 转义自动还原 - 升级流程：下载 PK 魔数检查与诊断日志；SHA 校验；替换改「改名腾位」绕开 Windows 文件锁延迟； nssm stop/start 驱动停止与拉起（pid 强杀回退）；升级中标志暂停看门狗拉起避免抢占 jar 锁 - ws_client：重连改循环实现修复递归栈溢出（长时间断线后 maximum recursion depth）； 退避封顶延长至 5 分钟；心跳线程独立修复「暂无心跳数据」 - 新增命令：GET_APIS/HTTP_PROXY（接口测试代理，纯透传鉴权）、GET_LOG_DOWNLOAD（日志下载备用） - 诊断增强：taskkill/进程退出/替换重试逐次日志 Co-Authored-By: Claude <noreply@anthropic.com>
195 lines
7.9 KiB
Python
195 lines
7.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Agent WebSocket 客户端。
|
||
|
||
- 主动出站连接 B 端（穿透白名单，无需 A 端开放入站端口）
|
||
- 断线指数退避重连（1s → 2s → 4s → ... 最大 60s）
|
||
- 周期心跳：拉取主程序 /api/admin/health 并转发给 B 端
|
||
- 收到下行命令交给 CommandDispatcher 处理
|
||
|
||
上行消息格式：{"type":"HEARTBEAT","deviceId":"...","ts":...,"payload":{...}}
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import threading
|
||
import time
|
||
|
||
import websocket # websocket-client
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
class AgentClient:
|
||
"""WebSocket 长连接客户端（连接管理 + 心跳 + 重连）。"""
|
||
|
||
def __init__(self, config, dispatcher):
|
||
self.config = config
|
||
self.dispatcher = dispatcher
|
||
|
||
self._ws = None # 当前 WebSocket 连接
|
||
self._connected = False # 是否已连接
|
||
self._reconnect_attempts = 0 # 连续重连次数（成功后清零）
|
||
self._lock = threading.Lock() # 保护 _ws/_connected 的并发访问
|
||
self._heartbeat_stop = threading.Event()
|
||
self._threads: list[threading.Thread] = []
|
||
|
||
# ================= 生命周期 =================
|
||
|
||
def start(self):
|
||
"""启动：连接循环线程 + 心跳线程。
|
||
|
||
连接线程跑 _connect_loop（循环重连，非递归——递归实现会在长时间断线后
|
||
撑爆 Python 调用栈，出现 maximum recursion depth exceeded）。
|
||
首次心跳立即发送，避免网关空闲超时在首个周期内掐断连接。
|
||
"""
|
||
t_heartbeat = threading.Thread(target=self._heartbeat_loop, name="heartbeat", daemon=True)
|
||
t_heartbeat.start()
|
||
self._threads.append(t_heartbeat)
|
||
t_connect = threading.Thread(target=self._connect_loop, name="ws-connect", daemon=True)
|
||
t_connect.start()
|
||
self._threads.append(t_connect)
|
||
|
||
def stop(self):
|
||
"""停止并释放资源。"""
|
||
self._heartbeat_stop.set()
|
||
with self._lock:
|
||
if self._ws is not None:
|
||
try:
|
||
self._ws.close()
|
||
except Exception:
|
||
pass
|
||
self._ws = None
|
||
self._connected = False
|
||
|
||
# ================= 连接管理 =================
|
||
|
||
def _connect_loop(self):
|
||
"""连接循环：断线后指数退避重连（循环代替递归，避免长时间断线栈溢出）。"""
|
||
while not self._heartbeat_stop.is_set():
|
||
try:
|
||
self._connect_once()
|
||
except Exception as e:
|
||
log.warning("连接过程异常：%s", e)
|
||
# 指数退避：1s 起翻倍至 5 分钟封顶（连接成功后 on_open 清零计数）
|
||
self._reconnect_attempts += 1
|
||
attempt = self._reconnect_attempts
|
||
delay = min(1000 * (2 ** min(attempt - 1, 9)), 300_000) / 1000.0
|
||
log.info("%.0fs 后进行第 %d 次重连", delay, attempt)
|
||
if self._heartbeat_stop.wait(delay):
|
||
return
|
||
|
||
def _connect_once(self):
|
||
"""建立一次 WebSocket 连接（阻塞直到连接断开；重连由 _connect_loop 负责）。"""
|
||
# 优先用网关基础地址推导：gateway.url(http→ws) + socket.path + 设备号
|
||
# 未配置 gateway.url 时回退 server.url（以 /agentSocket/ 结尾）+ 设备号
|
||
if self.config.gateway_url:
|
||
ws_base = self.config.gateway_url.rstrip("/")
|
||
if ws_base.startswith("http://"):
|
||
ws_base = "ws://" + ws_base[len("http://"):]
|
||
elif ws_base.startswith("https://"):
|
||
ws_base = "wss://" + ws_base[len("https://"):]
|
||
url = ws_base + "/" + self.config.socket_path.strip("/") + "/" + self.config.device_id
|
||
else:
|
||
url = self.config.server_url.rstrip("/") + "/" + self.config.device_id
|
||
log.info("连接管理服务器：%s", url)
|
||
|
||
headers = [
|
||
f"X-Device-Id: {self.config.device_id}",
|
||
f"X-Device-Token: {self.config.device_token}",
|
||
]
|
||
|
||
def on_open(ws):
|
||
self._connected = True
|
||
self._reconnect_attempts = 0
|
||
log.info("已连接管理服务器")
|
||
self.send("REGISTER", {
|
||
"deviceId": self.config.device_id,
|
||
"version": "1.0.0",
|
||
})
|
||
|
||
def on_message(ws, text):
|
||
try:
|
||
node = json.loads(text)
|
||
msg_type = node.get("type")
|
||
payload = node.get("payload")
|
||
self.dispatcher.dispatch(msg_type, payload)
|
||
except Exception as e:
|
||
log.warning("命令处理异常：%s", e)
|
||
|
||
def on_close(ws, code, reason):
|
||
self._connected = False
|
||
# 区分失败类型：1000/1001 为正常关闭；其他关闭码（如 1008 策略拒绝）多为设备未注册或 token 错误
|
||
# 重连由 _connect_loop 统一调度（on_close 不再递归调度）
|
||
if code in (1000, 1001):
|
||
log.info("连接关闭：code=%s, reason=%s", code, reason)
|
||
elif reason:
|
||
log.warning("连接被服务器拒绝（设备未注册或 token 错误）：code=%s, reason=%s", code, reason)
|
||
else:
|
||
log.warning("连接被服务器关闭：code=%s（设备未注册或 token 错误）", code)
|
||
|
||
def on_error(ws, error):
|
||
self._connected = False
|
||
# websocket-client 收到服务器关闭帧时会先回调本方法（特征：fin=1 opcode=8），
|
||
# 这不是网络问题，而是被服务器拒绝（设备未注册 / token 错误）
|
||
if "opcode=8" in str(error):
|
||
log.warning("连接被服务器拒绝（设备未注册或 token 错误）：%s", error)
|
||
else:
|
||
log.warning("网络不通（无法连接管理服务器）：%s，将重连", error)
|
||
# websocket-client 出错后必定回调 on_close，重连由 on_close 统一调度
|
||
# 此处不重复调度，避免双重重连
|
||
|
||
ws = websocket.WebSocketApp(
|
||
url,
|
||
header=headers,
|
||
on_open=on_open,
|
||
on_message=on_message,
|
||
on_close=on_close,
|
||
on_error=on_error,
|
||
)
|
||
with self._lock:
|
||
self._ws = ws
|
||
# 阻塞运行（跳过协议层 ping，应用层心跳已足够；关掉默认 ping 避免双重心跳）
|
||
ws.run_forever(ping_interval=25, ping_timeout=10)
|
||
|
||
# ================= 心跳 =================
|
||
|
||
def _heartbeat_loop(self):
|
||
"""周期心跳：拉主程序健康数据转发。"""
|
||
while not self._heartbeat_stop.wait(self.config.heartbeat_sec):
|
||
try:
|
||
self._heartbeat_once()
|
||
except Exception as e:
|
||
log.warning("心跳发送异常：%s", e)
|
||
|
||
def _heartbeat_once(self):
|
||
main_health = self.dispatcher.main_app_client.health()
|
||
self.send("HEARTBEAT", {
|
||
"deviceId": self.config.device_id,
|
||
"mainApp": main_health,
|
||
})
|
||
|
||
# ================= 上行消息 =================
|
||
|
||
def send(self, msg_type: str, payload):
|
||
"""发送上行消息（未连接时丢弃并记日志）。"""
|
||
with self._lock:
|
||
ws = self._ws
|
||
connected = self._connected
|
||
if ws is None or not connected:
|
||
log.debug("未连接，丢弃上行消息：%s", msg_type)
|
||
return
|
||
envelope = {
|
||
"type": msg_type,
|
||
"deviceId": self.config.device_id,
|
||
"ts": int(time.time()),
|
||
"payload": payload,
|
||
}
|
||
try:
|
||
ws.send(json.dumps(envelope, ensure_ascii=False))
|
||
except Exception as e:
|
||
log.warning("上行消息发送失败（%s）：%s", msg_type, e)
|
||
|
||
def is_connected(self) -> bool:
|
||
"""当前是否已连接（供健康状态展示）。"""
|
||
return self._connected
|