From 7e4236e2f3c3b6468cbb59661b9876e215a1330f Mon Sep 17 00:00:00 2001 From: CNWei Date: Sat, 1 Aug 2026 18:58:12 +0800 Subject: [PATCH] fix: address unimplemented/broken issues across server and desktop --- desktop/lib/src/terminal_screen.dart | 6 +- desktop/test/e2e_ws_test.dart | 10 +- desktop/test/regression_ws_test.dart | 77 +++++++++++++++ desktop/test/support.dart | 14 +++ src/config.rs | 8 ++ src/terminal/engine.rs | 11 ++- src/ws/handler.rs | 137 +++++++++++++++++++++++---- 7 files changed, 239 insertions(+), 24 deletions(-) create mode 100644 desktop/test/regression_ws_test.dart create mode 100644 desktop/test/support.dart diff --git a/desktop/lib/src/terminal_screen.dart b/desktop/lib/src/terminal_screen.dart index 3922a7f..0c9ee7a 100644 --- a/desktop/lib/src/terminal_screen.dart +++ b/desktop/lib/src/terminal_screen.dart @@ -320,12 +320,12 @@ class _StatusBar extends StatelessWidget { ), ), SizedBox(width: 20), - _Hint('鼠标拖选复制', RttyTheme.textFaint), + _Hint('拖选复制', RttyTheme.textFaint), SizedBox(width: 14), - _Hint('右键粘贴', RttyTheme.textFaint), + _Hint('Ctrl+Shift+C / V 复制粘贴', RttyTheme.textFaint), Spacer(), Text( - 'FLUTTER × XTERM × ALCATTERM ENGINE', + 'FLUTTER × XTERM TERMINAL', style: TextStyle(color: RttyTheme.textFaint, fontSize: 10), ), ], diff --git a/desktop/test/e2e_ws_test.dart b/desktop/test/e2e_ws_test.dart index b3ae2b4..53dc01b 100644 --- a/desktop/test/e2e_ws_test.dart +++ b/desktop/test/e2e_ws_test.dart @@ -2,18 +2,22 @@ // // 运行前需先启动 Rust 服务端：cd .. && cargo run // 运行：flutter test test/e2e_ws_test.dart -import 'dart:io'; - import 'package:flutter_test/flutter_test.dart'; import 'package:xterm/xterm.dart'; -import '../lib/src/rtty_client.dart'; +import 'package:rtty_desktop/src/rtty_client.dart'; + +import 'support.dart'; void main() { TestWidgetsFlutterBinding.ensureInitialized(); test('RttyClient connects to Rust server and renders command output', () async { + if (!await isServerUp('127.0.0.1', 8080)) { + markTestSkipped('Rtty server not running'); + return; + } final terminal = Terminal(); final client = RttyClient(terminal); diff --git a/desktop/test/regression_ws_test.dart b/desktop/test/regression_ws_test.dart new file mode 100644 index 0000000..dabbabb --- /dev/null +++ b/desktop/test/regression_ws_test.dart @@ -0,0 +1,77 @@ +// 多端解耦 + 控制权强制回归验证（连真实 Rust 服务端）。 +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:web_socket_channel/io.dart'; + +import 'support.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + test('多端解耦 + 控制权强制 + 会话清理', () async { + if (!await isServerUp('127.0.0.1', 8080)) { + markTestSkipped('Rtty server not running'); + return; + } + final base = 'ws://127.0.0.1:8080/ws'; + + // 1. 桌面端（client=desktop）：收 raw 二进制，不收 snapshot。 + final desktop = IOWebSocketChannel.connect(Uri.parse('$base?client=desktop')); + final dOut = []; + desktop.stream.listen((m) { + if (m is List) { + dOut.add('binary:${utf8.decode(m, allowMalformed: true)}'); + } else if (m is String) { + dOut.add('text:${m.length > 100 ? m.substring(0, 100) : m}'); + } else { + dOut.add('$m'); + } + }); + await Future.delayed(const Duration(milliseconds: 700)); + final dHasBinary = dOut.any((e) => e.startsWith('binary:')); + final dHasSnapshot = dOut.any((e) => e.contains('mobile_snapshot')); + expect(dHasBinary, isTrue, reason: '桌面端应收到原始 ANSI 二进制 $dOut'); + expect(dHasSnapshot, isFalse, reason: '桌面端不应收到语义快照 $dOut'); + + // 2. 移动端（client=mobile）：收 snapshot，不收 binary。 + final mobile = IOWebSocketChannel.connect(Uri.parse('$base?client=mobile')); + final mOut = []; + mobile.stream.listen((m) { + if (m is List) { + mOut.add('binary:${utf8.decode(m, allowMalformed: true)}'); + } else if (m is String) { + mOut.add('text:${m.length > 100 ? m.substring(0, 100) : m}'); + } else { + mOut.add('$m'); + } + }); + await Future.delayed(const Duration(milliseconds: 700)); + final mHasSnapshot = mOut.any((e) => e.contains('mobile_snapshot')); + final mHasBinary = mOut.any((e) => e.startsWith('binary:')); + expect(mHasSnapshot, isTrue, reason: '移动端应收到语义快照 $mOut'); + expect(mHasBinary, isFalse, reason: '移动端不应收到二进制 $mOut'); + + // 3. 桌面端发命令，确认 raw 输出回显。 + desktop.sink.add(jsonEncode({'type': 'input', 'data': 'echo D_E2E\r\n'})); + await Future.delayed(const Duration(milliseconds: 900)); + expect(dOut.join('\n').contains('D_E2E'), isTrue, reason: '桌面端命令回显'); + + // 4. 控制权强制：桌面端 claim 后，移动端输入被拒。 + desktop.sink.add(jsonEncode({'type': 'claim_control'})); + await Future.delayed(const Duration(milliseconds: 400)); + mobile.sink + .add(jsonEncode({'type': 'input', 'data': 'echo BLOCKED_SHOULD_NOT_SHOW\r\n'})); + await Future.delayed(const Duration(milliseconds: 900)); + final both = (dOut.join('\n') + mOut.join('\n')); + expect(both.contains('BLOCKED_SHOULD_NOT_SHOW'), isFalse, + reason: '非控制者输入应被阻止'); + + // 5. 会话清理：exit 触发会话移除。 + desktop.sink.add(jsonEncode({'type': 'input', 'data': 'exit\r\n'})); + await Future.delayed(const Duration(milliseconds: 1300)); + + desktop.sink.close(); + mobile.sink.close(); + }); +} diff --git a/desktop/test/support.dart b/desktop/test/support.dart new file mode 100644 index 0000000..25477b5 --- /dev/null +++ b/desktop/test/support.dart @@ -0,0 +1,14 @@ +// 集成测试共享工具。 +import 'package:web_socket_channel/io.dart'; + +/// 探测 Rtty 服务端是否可达；返回 false 时集成测试应被跳过。 +Future isServerUp(String host, int port) async { + try { + final ws = IOWebSocketChannel.connect(Uri.parse('ws://$host:$port/ws')); + await ws.ready.timeout(const Duration(seconds: 2)); + ws.sink.close(); + return true; + } catch (_) { + return false; + } +} diff --git a/src/config.rs b/src/config.rs index d79dbd0..242e253 100644 --- a/src/config.rs +++ b/src/config.rs @@ -22,6 +22,8 @@ pub struct ServerConfig { pub max_scrollback: usize, /// 单个会话允许的最大并发客户端数（0 表示不限制）。 pub max_clients: usize, + /// 会话在无客户端连接后保留的秒数；超时则清理（支持断线重连窗口）。 + pub idle_timeout_secs: u64, } impl Default for ServerConfig { @@ -34,6 +36,7 @@ impl Default for ServerConfig { rows: 32, max_scrollback: 10_000, max_clients: 16, + idle_timeout_secs: 60, } } } @@ -74,6 +77,11 @@ impl ServerConfig { { cfg.max_clients = n; } + if let Ok(v) = env::var("RTTY_IDLE_TIMEOUT") + && let Ok(n) = v.parse() + { + cfg.idle_timeout_secs = n; + } cfg } diff --git a/src/terminal/engine.rs b/src/terminal/engine.rs index 0be9e43..32a5e04 100644 --- a/src/terminal/engine.rs +++ b/src/terminal/engine.rs @@ -71,9 +71,16 @@ pub struct TerminalEngine { impl TerminalEngine { /// 创建一个新的终端引擎。 - pub fn new(size: TermSize, writer: Arc>>) -> Self { + /// + /// `scrollback` 为滚动历史的行数上限，对应 alacritty 的 `scrolling_history`。 + pub fn new( + size: TermSize, + writer: Arc>>, + scrollback: usize, + ) -> Self { let listener = SessionListener::new(writer); - let term = Term::new(Config::default(), &size, listener); + let config = Config { scrolling_history: scrollback, ..Config::default() }; + let term = Term::new(config, &size, listener); Self { term, parser: Processor::new() } } diff --git a/src/ws/handler.rs b/src/ws/handler.rs index 0ce0399..3efc2ab 100644 --- a/src/ws/handler.rs +++ b/src/ws/handler.rs @@ -25,6 +25,25 @@ static SESSION_SEQ: AtomicU64 = AtomicU64::new(0); /// 客户端全局 ID 计数器。 static CLIENT_SEQ: AtomicU64 = AtomicU64::new(0); +/// 客户端类型：决定它订阅哪一路输出流（多端解耦）。 +#[derive(Clone, Copy, PartialEq, Eq)] +enum ClientKind { + /// 桌面端：订阅原始 ANSI 二进制流。 + Desktop, + /// 移动端：订阅语义化 JSON 快照。 + Mobile, +} + +impl ClientKind { + /// 从 WebSocket 查询参数解析客户端类型，默认桌面端。 + fn from_params(params: &HashMap) -> Self { + match params.get("client").map(|s| s.as_str()) { + Some("mobile") => Self::Mobile, + _ => Self::Desktop, + } + } +} + /// 广播给订阅者的输出事件。 #[derive(Clone)] pub enum OutputEvent { @@ -67,6 +86,7 @@ async fn handle_socket( params: HashMap, ) { let client_id = format!("client-{}", CLIENT_SEQ.fetch_add(1, Ordering::Relaxed)); + let kind = ClientKind::from_params(¶ms); let requested = params.get("session").cloned().unwrap_or_default(); let session = match get_or_create_session(&state, &requested) { @@ -87,10 +107,22 @@ async fn handle_socket( return; } - // 发送当前快照（断线重连 / 新加入状态恢复）。 - let snap = session.engine.lock().unwrap().snapshot(); - let init = ServerMessage::MobileSnapshot { data: snap }; - if tx.send(Message::Text(init.to_json())).await.is_err() { + // 移动端在连接时获取一次语义化快照作为初始状态（桌面端走原始 ANSI 流）。 + if kind == ClientKind::Mobile { + let snap = session.engine.lock().unwrap().snapshot(); + let init = ServerMessage::MobileSnapshot { data: snap }; + if tx.send(Message::Text(init.to_json())).await.is_err() { + return; + } + } + + // 订阅前检查并发上限。 + let max_clients = state.config.max_clients; + if max_clients > 0 && session.clients.load(Ordering::SeqCst) >= max_clients { + let msg = ServerMessage::Error { + message: format!("session {} is at max client capacity", session.id), + }; + let _ = tx.send(Message::Text(msg.to_json())).await; return; } @@ -100,18 +132,22 @@ async fn handle_socket( loop { tokio::select! { - // 服务端输出 -> 客户端 + // 服务端输出 -> 客户端（按客户端类型过滤，实现多端解耦）。 out = out_rx.recv() => { match out { Ok(OutputEvent::Raw(bytes)) => { - if tx.send(Message::Binary(bytes)).await.is_err() { + if kind == ClientKind::Desktop + && tx.send(Message::Binary(bytes)).await.is_err() + { break; } } Ok(OutputEvent::Snapshot(s)) => { - let msg = ServerMessage::MobileSnapshot { data: s }; - if tx.send(Message::Text(msg.to_json())).await.is_err() { - break; + if kind == ClientKind::Mobile { + let msg = ServerMessage::MobileSnapshot { data: s }; + if tx.send(Message::Text(msg.to_json())).await.is_err() { + break; + } } } Ok(OutputEvent::Exit) => { @@ -143,6 +179,23 @@ async fn handle_socket( } session.clients.fetch_sub(1, Ordering::SeqCst); + + // 若已无客户端，安排空闲超时清理（兜底机制，Windows ConPTY 下进程退出 + // 检测不可靠，依赖超时确保会话最终被释放）。 + if session.clients.load(Ordering::SeqCst) == 0 { + spawn_idle_cleanup(session, state); + } +} + +/// 空闲超时清理：若无客户端重连，则终止会话并释放资源。 +fn spawn_idle_cleanup(session: Arc, state: Arc) { + let timeout = state.config.idle_timeout_secs; + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(timeout)).await; + if session.clients.load(Ordering::SeqCst) == 0 { + cleanup_session(&session, &state); + } + }); } /// 处理单条客户端 JSON 消息，返回需要回发给该客户端的 JSON（若无则 None）。 @@ -158,7 +211,15 @@ fn handle_client_message( match msg { ClientMessage::Input { data } => { - let _ = session.pty.lock().unwrap().write(data.as_bytes()); + // 控制权强制：若存在控制者且不是当前客户端，则拒绝写入， + // 避免多端同时输入互相干扰。 + let blocked = { + let control = session.control.lock().unwrap(); + control.as_deref().is_some_and(|h| h != client_id) + }; + if !blocked { + let _ = session.pty.lock().unwrap().write(data.as_bytes()); + } None } ClientMessage::Resize { cols, rows } => { @@ -185,7 +246,7 @@ fn handle_client_message( /// 获取或创建会话。 fn get_or_create_session( - state: &crate::AppState, + state: &Arc, requested: &str, ) -> Result> { if !requested.is_empty() @@ -195,18 +256,23 @@ fn get_or_create_session( } let id = format!("session-{}", SESSION_SEQ.fetch_add(1, Ordering::Relaxed)); - let session = create_session(id.clone(), &state.config)?; + let session = create_session(id.clone(), &state.config, state.clone())?; state.sessions.insert(id.clone(), session.clone()); Ok(session) } /// 创建会话并启动 PTY 读取任务。 -fn create_session(id: String, config: &ServerConfig) -> Result> { +fn create_session( + id: String, + config: &ServerConfig, + state: Arc, +) -> Result> { let mut pty = PtySession::new(&config.shell, config.cols, config.rows)?; let writer = pty.writer().clone(); let engine = Arc::new(Mutex::new(TerminalEngine::new( TermSize { columns: config.cols as usize, rows: config.rows as usize }, writer, + config.max_scrollback, ))); let reader = pty.take_reader().ok_or_else(|| anyhow::anyhow!("pty reader already taken"))?; @@ -220,12 +286,22 @@ fn create_session(id: String, config: &ServerConfig) -> Result> { clients: AtomicUsize::new(0), }); - start_pty_reader(session.clone(), reader); + start_pty_reader(session.clone(), reader, state); Ok(session) } /// 启动阻塞的 PTY 读取任务，读取输出并广播。 -fn start_pty_reader(session: Arc, mut reader: Box) { +/// +/// 同时启动一个进程监控任务，通过轮询 `try_wait` 检测子进程退出—— +/// 因为 Windows ConPTY 下 shell 退出可能不触发 PTY EOF，仅靠 EOF 无法可靠清理。 +fn start_pty_reader( + session: Arc, + mut reader: Box, + state: Arc, +) { + // 进程退出监控：一旦子进程退出即清理会话。 + spawn_exit_monitor(session.clone(), state.clone()); + tokio::task::spawn_blocking(move || { let mut buf = vec![0u8; 8192]; loop { @@ -251,10 +327,39 @@ fn start_pty_reader(session: Arc, mut reader: Box) { let _ = session.output.send(OutputEvent::Snapshot(snapshot)); } - let _ = session.output.send(OutputEvent::Exit); + // PTY 读到 EOF（Unix 场景）：触发会话结束并清理。 + cleanup_session(&session, &state); }); } +/// 轮询子进程退出状态，退出后清理会话。 +fn spawn_exit_monitor(session: Arc, state: Arc) { + tokio::task::spawn_blocking(move || { + // 最多监控 5 分钟，避免无意义长驻。 + for _ in 0..3000 { + let exited = match session.pty.lock() { + Ok(mut pty) => pty.try_wait().map(|s| s.is_some()).unwrap_or(false), + Err(_) => false, + }; + if exited { + cleanup_session(&session, &state); + return; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + }); +} + +/// 触发会话结束事件并清理资源（幂等）。 +fn cleanup_session(session: &Session, state: &crate::AppState) { + let _ = session.output.send(OutputEvent::Exit); + if let Ok(mut pty) = session.pty.lock() { + let _ = pty.kill(); + } + state.sessions.remove(&session.id); + tracing::info!("session {} closed", session.id); +} + /// 发送一条文本消息，忽略错误。 async fn send_text(socket: WebSocket, text: String) -> Result<()> { let (mut tx, _rx) = socket.split();