"""
pytcper TCP调试助手 - TCP 服务端标签页
======================================
本地监听端口，接收多个客户端连接，支持定向发送、广播、主动断开客户端，
客户端列表按名称匹配、每秒刷新时保留选中。
"""
import socket
import threading
import time
import tkinter as tk
from tkinter import messagebox, ttk
from base_tab import _BaseTab
from tcp_utils import FONT, SOCK_TIMEOUT, close_socket, fmt_duration
class ServerTab(_BaseTab):
"""TCP 服务端：本地监听端口，接收多个客户端连接并收发指令。"""
def __init__(self, master):
super().__init__(master, build_top=self._build_top)
ttk.Button(self.send_row, text="广播", command=self.do_broadcast).pack(side="right", padx=4)
self._uptime_after = self.after(1000, self._tick_uptime)
def _build_top(self):
top = ttk.Frame(self)
top.pack(fill="x", padx=8, pady=(8, 2))
ttk.Label(top, text="监听端口:").pack(side="left")
self.var_port = tk.StringVar(value="8080")
ttk.Entry(top, textvariable=self.var_port, width=8).pack(side="left", padx=(2, 8))
self.start_btn = ttk.Button(top, text="启动服务", command=self.toggle_server)
self.start_btn.pack(side="left", padx=(4, 8))
self.srv_state = tk.StringVar(value="未启动")
ttk.Label(top, textvariable=self.srv_state).pack(side="left")
cli_frame = ttk.LabelFrame(self, text="已连接客户端（单击选中发送对象）")
cli_frame.pack(fill="x", padx=8, pady=2)
self.client_listbox = tk.Listbox(cli_frame, height=3, font=FONT)
self.client_listbox.pack(side="left", fill="both", expand=True, padx=4, pady=4)
btn_col = ttk.Frame(cli_frame)
btn_col.pack(side="left", fill="y", padx=(0, 4), pady=4)
ttk.Button(btn_col, text="刷新", command=self._refresh_client_list).pack(fill="x")
ttk.Button(btn_col, text="断开所选", command=self._kick_selected).pack(fill="x", pady=(2, 0))
def toggle_server(self):
if self.server_sock is not None:
self._stop_server()
return
try:
port = int(self.var_port.get().strip())
except ValueError:
messagebox.showerror("错误", "端口必须是数字")
return
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.settimeout(1.0)
try:
srv.bind(("0.0.0.0", port))
srv.listen(16)
except OSError as exc:
srv.close()
self._sys_log("启动服务失败: %s" % exc)
messagebox.showerror("启动失败", str(exc))
return
self.server_sock = srv
self.start_btn.configure(text="停止服务")
self.srv_state.set("监听中 0.0.0.0:%d" % port)
self._sys_log("服务已启动，监听端口 %d（本机地址 127.0.0.1:%d）" % (port, port))
threading.Thread(target=self._accept_loop, daemon=True).start()
def _stop_server(self):
srv, self.server_sock = self.server_sock, None
close_socket(srv)
with self._lock:
socks = list(self.clients.keys())
for sock in socks:
self._remove_client(sock)
self.start_btn.configure(text="启动服务")
self.srv_state.set("未启动")
self._sys_log("服务已停止")
def _enable_keepalive(self, sock):
"""开启 TCP keepalive，让 OS 底层检测死连接。
客户端拔网线 / 进程崩溃（无 FIN / RST）时，recv 会一直
timeout→continue 而永不返回；开启 keepalive 后 OS 会在
对端无响应时最终使 recv 抛错，从而及时调用 _remove_client。
"""
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
try:
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 30)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)
except OSError:
pass
except OSError:
pass
def _accept_loop(self):
while True:
srv = self.server_sock
if srv is None:
break
try:
sock, addr = srv.accept()
except socket.timeout:
continue
except OSError:
break
self._enable_keepalive(sock)
sock.settimeout(SOCK_TIMEOUT)
name = "%s:%d" % addr
with self._lock:
self.clients[sock] = name
self.client_list.append(sock)
self.client_times[sock] = time.monotonic()
self._sys_log("客户端 %s 已连接" % name)
self.msg_queue.put(("clients", None))
threading.Thread(target=self._client_recv_loop, args=(sock, name), daemon=True).start()
def _client_recv_loop(self, sock, name):
dec_state = None
while self.server_sock is not None:
try:
data = sock.recv(4096)
except socket.timeout:
continue
except OSError:
break
if not data:
break
text, dec_state = self._decode_stream(dec_state, data)
self._bump_stats("rx", data)
self._log("rx", text, name)
self._remove_client(sock)
def _remove_client(self, sock):
with self._lock:
if sock not in self.clients:
return
name = self.clients.pop(sock)
if sock in self.client_list:
self.client_list.remove(sock)
self.client_times.pop(sock, None)
close_socket(sock)
self._sys_log("客户端 %s 已断开" % name)
self.msg_queue.put(("clients", None))
def _kick_selected(self):
"""断开列表中选中的客户端。"""
target = self._selected_client()
if target is None:
messagebox.showwarning("提示", "请先选中要断开的客户端（或该客户端已断开）")
return
sock, _name = target
close_socket(sock)
self._remove_client(sock)
def _selected_client(self):
"""解析列表选中行对应的客户端，返回 (sock, name)；无效返回 None。
列表行格式为 "ip:port 在线 HH:MM:SS"，行与 socket 之间用名称匹配
而非索引，避免客户端刚断开、列表尚未刷新时索引错位发错对象。
"""
sel = self.client_listbox.curselection()
if not sel:
return None
try:
line = self.client_listbox.get(sel[0])
except tk.TclError:
return None
name = line.split(" 在线 ", 1)[0]
with self._lock:
candidates = [s for s, n in self.clients.items() if n == name]
if not candidates:
return None
sock = max(candidates, key=lambda s: self.client_times.get(s, 0.0))
return sock, name
def _refresh_client_list(self):
with self._lock:
items = [(self.clients[s], time.monotonic() - t)
for s, t in self.client_times.items() if s in self.clients]
selected_name = None
sel = self.client_listbox.curselection()
if sel:
try:
selected_name = self.client_listbox.get(sel[0]).split(" 在线 ", 1)[0]
except tk.TclError:
pass
self.client_listbox.delete(0, "end")
for name, up in items:
self.client_listbox.insert("end", "%s 在线 %s" % (name, fmt_duration(up)))
if selected_name is not None:
for idx, (name, _up) in enumerate(items):
if name == selected_name:
self.client_listbox.selection_set(idx)
self.client_listbox.see(idx)
break
def _tick_uptime(self):
if self.server_sock is not None and not self._closed:
self._refresh_client_list()
if not self._closed:
self._uptime_after = self.after(1000, self._tick_uptime)
def _send_to(self, sock, name, data):
exc = self._sendall_safe(sock, data)
if exc is not None:
self._sys_log("发送给 %s 失败: %s" % (name, exc))
return False
self._log("tx", self._tx_text(data), "→" + name)
self._bump_stats("tx", data)
return True
def do_send(self, silent=False):
prepared = self._prepare_send(silent)
if prepared is None:
return False
raw, data = prepared
target = self._selected_client()
if target is None:
if not silent:
messagebox.showwarning("提示", "请先在列表中选中目标客户端（或该客户端已断开）")
return False
sock, name = target
if not silent:
self._record_history(raw)
return self._send_to(sock, name, data)
def do_broadcast(self, silent=False):
prepared = self._prepare_send(silent)
if prepared is None:
return False
raw, data = prepared
with self._lock:
targets = list(self.clients.items())
if not targets:
if not silent:
messagebox.showwarning("提示", "当前没有已连接的客户端")
return False
ok = True
for sock, name in targets:
ok = self._send_to(sock, name, data) and ok
self._sys_log("已广播给 %d 个客户端" % len(targets))
return ok
def _send_raw(self, data, desc):
target = self._selected_client()
if target is None:
messagebox.showwarning("提示", "请先在列表中选中目标客户端（或该客户端已断开）")
return False
return self._send_to(target[0], target[1], data)
def _handle_msg(self, kind, payload):
if kind == "clients":
self._refresh_client_list()
def _close_conns(self):
if self._uptime_after is not None:
try:
self.after_cancel(self._uptime_after)
except Exception:
pass
self._uptime_after = None
super()._close_conns()
srv, self.server_sock = self.server_sock, None
if srv is not None:
close_socket(srv)
with self._lock:
socks = list(self.clients.keys())
for sock in socks:
self._remove_client(sock)