From 06796031e6cef60d4e634df33dbab50399742183 Mon Sep 17 00:00:00 2001 From: TomZz Date: Fri, 12 Jun 2026 00:26:04 +0800 Subject: [PATCH] Refactor SFTP transfers with concurrency, persistence, and UI improvements MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用多线程并发处理多文件/大文件 SFTP 传输以提升速度 - 修复 SFTP 底部状态栏不显示的排版问题 - 新增持久化功能：将传输历史记录保存至本地配置文件夹并在启动时加载 - 修复传输记录中文件和文件夹数量统计及 i18n 汉化的正确显示 - 优化传输记录窗口 UI：增加自适应滚动列表，并集成匹配 SFTP 面板原生样式的右侧长驻滚动条 - 清理 Cargo warning，修复无用变量等问题 --- Cargo.lock | 1 + Cargo.toml | 1 + locales/en.yml | 12 ++ locales/zh-CN.yml | 12 ++ src/config.rs | 13 ++ src/main.rs | 470 +++++++++++++++++++++++++++++++++--------- src/sftp.rs | 509 +++++++++++++++++++++++++++++++++++----------- src/terminal.rs | 50 ++++- 8 files changed, 860 insertions(+), 208 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6429233..ba7a17b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -292,6 +292,7 @@ dependencies = [ "chrono", "directories", "flate2", + "futures", "gpui", "gpui-component", "gpui-component-assets", diff --git a/Cargo.toml b/Cargo.toml index 78396e9..78e66d4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,7 @@ uuid = { version = "1", features = ["v4", "serde"] } walkdir = "2" zip = { version = "2", default-features = false, features = ["deflate"] } image = { version = "0.25.9", default-features = false, features = ["png"] } +futures = "0.3.32" [profile.release] opt-level = 3 diff --git a/locales/en.yml b/locales/en.yml index 40021c3..0523415 100644 --- a/locales/en.yml +++ b/locales/en.yml @@ -71,6 +71,18 @@ uploaded_folder: "Uploaded folder" uploaded_n_files: "Uploaded %{files} files" uploaded_n_folders: "Uploaded %{folders} folders" uploaded_files_and_folders: "Uploaded %{files} files and %{folders} folders" +n_files: "%{files} files" +n_folders: "%{folders} folders" +n_files_and_folders: "%{files} files and %{folders} folders" reset: "Reset" reset_layout: "Reset Default Layout" reset_layout_success: "Layout reset. Window size takes effect on restart." + +transfers: "Transfers" +no_transfers_yet: "No transfers yet" +downloading: "Downloading" +paused: "Paused" +completed: "Completed" +failed: "Failed" +cancelled: "Cancelled" +session: "Session" diff --git a/locales/zh-CN.yml b/locales/zh-CN.yml index fb81d46..d2b865a 100644 --- a/locales/zh-CN.yml +++ b/locales/zh-CN.yml @@ -71,6 +71,18 @@ uploaded_folder: "已上传文件夹" uploaded_n_files: "已上传 %{files} 个文件" uploaded_n_folders: "已上传 %{folders} 个文件夹" uploaded_files_and_folders: "已上传 %{files} 个文件及 %{folders} 个文件夹" +n_files: "%{files} 个文件" +n_folders: "%{folders} 个文件夹" +n_files_and_folders: "%{files} 个文件及 %{folders} 个文件夹" reset: "重置" reset_layout: "恢复默认布局" reset_layout_success: "布局占比已恢复默认。窗口大小将在下次启动时生效。" + +transfers: "传输记录" +no_transfers_yet: "暂无传输记录" +downloading: "正在下载" +paused: "已暂停" +completed: "已完成" +failed: "失败" +cancelled: "已取消" +session: "会话" diff --git a/src/config.rs b/src/config.rs index 869057a..c8f647e 100644 --- a/src/config.rs +++ b/src/config.rs @@ -115,6 +115,8 @@ pub struct ConfigFile { pub workspace_panels: Option>, #[serde(default)] pub body_panels: Option>, + #[serde(default)] + pub transfers: Vec, } fn default_locale() -> String { @@ -230,6 +232,17 @@ impl ConfigStore { self.cache.body_panels.as_ref() } + pub fn transfers(&self) -> Vec { + self.cache.transfers.clone() + } + + pub fn set_transfers(&mut self, transfers: Vec) { + self.cache.transfers = transfers; + if let Err(err) = self.save() { + tracing::error!("failed to save config: {err:#}"); + } + } + pub fn set_layout_state( &mut self, window_bounds: Option, diff --git a/src/main.rs b/src/main.rs index 9e22842..9d3cbb9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,7 +18,7 @@ use gpui::{ Focusable as _, FontWeight, Hsla, InteractiveElement as _, IntoElement, KeyBinding, KeyDownEvent, MouseButton, MouseDownEvent, MouseMoveEvent, MouseUpEvent, ParentElement as _, PathPromptOptions, Pixels, Point, QuitMode, Render, ScrollDelta, ScrollWheelEvent, - SharedString, Size, Styled as _, UniformListScrollHandle, Window, WindowOptions, div, point, + SharedString, Size, StatefulInteractiveElement, Styled as _, UniformListScrollHandle, Window, WindowOptions, div, point, prelude::FluentBuilder as _, px, size, uniform_list, }; use gpui_component::{ @@ -133,6 +133,8 @@ impl ScrollbarHandle for TerminalScrollbarHandle { } } + + struct Ashell { focus_handle: FocusHandle, selector_focus_handle: FocusHandle, @@ -163,6 +165,8 @@ struct Ashell { pending_sftp_path_sync: Option, sftp_context_menu: Option, show_hidden_files: bool, + transfers: Vec, + show_transfers_dialog: bool, system_status: Option, terminal_bounds: Option>, terminal_selecting: bool, @@ -311,6 +315,8 @@ impl Ashell { pending_sftp_path_sync: Some("/".into()), sftp_context_menu: None, show_hidden_files: false, + transfers: config.transfers(), + show_transfers_dialog: false, system_status: None, terminal_bounds: None, terminal_selecting: false, @@ -381,6 +387,7 @@ impl Ashell { } fn drain_backend_events(&mut self) { + let mut transfers_changed = false; while let Ok(event) = self.events_rx.try_recv() { match event { BackendEvent::Output { tab_id, bytes } => { @@ -503,8 +510,46 @@ impl Ashell { } } } + BackendEvent::TransferStarted { tab_id, info } => { + let tab_title = self.tabs.iter().find(|t| t.id == tab_id).map(|t| t.title.clone()).unwrap_or_else(|| "Unknown".to_string()); + self.transfers.insert( + 0, + crate::terminal::Transfer { + tab_id, + tab_title, + info, + transferred: 0, + total: None, + state: crate::terminal::TransferState::Running, + }, + ); + if self.transfers.len() > 50 { + self.transfers.truncate(50); + } + self.show_transfers_dialog = true; + transfers_changed = true; + } + BackendEvent::TransferProgress { + tab_id: _, + id, + transferred, + total, + state, + } => { + if let Some(t) = self.transfers.iter_mut().find(|t| t.info.id == id) { + t.transferred = transferred; + if let Some(total) = total { + t.total = Some(total); + } + t.state = state; + transfers_changed = true; + } + } } } + if transfers_changed { + self.config.set_transfers(self.transfers.clone()); + } } fn sample_system_if_due(&mut self) { @@ -1030,6 +1075,225 @@ impl Ashell { }); } + fn show_transfers_dialog(&mut self, window: &mut Window, cx: &mut Context) { + let view = cx.entity(); + window.open_dialog(cx, move |dialog: Dialog, _window, _| { + dialog + .title(t!("transfers").to_string()) + .w(px(600.)) + .content({ + let view = view.clone(); + move |content, window, cx| { + let mut transfers = view.read(cx).transfers.clone(); + transfers.sort_by_key(|t| { + match t.state { + crate::terminal::TransferState::Running | crate::terminal::TransferState::Paused => 0, + _ => 1, + } + }); + + if transfers.is_empty() { + return content.child( + div() + .p_4() + .text_center() + .text_color(cx.theme().muted_foreground) + .child(t!("no_transfers_yet").to_string()), + ); + } + + let list = v_flex().gap_2().children(transfers.into_iter().map(|t| { + let (icon, _color) = match t.info.kind { + crate::terminal::TransferType::Upload => (IconName::ArrowUp, cx.theme().primary), + crate::terminal::TransferType::Download => (IconName::ArrowDown, cx.theme().success), + }; + + let (status_text, actions) = match t.state { + crate::terminal::TransferState::Running => { + let percent = t.total.map(|tot| (t.transferred as f64 / tot as f64 * 100.0).clamp(0.0, 100.0)).unwrap_or(0.0); + let txt = if let Some(tot) = t.total { + format!("{:.1}% ({}/{})", percent, format_bytes(t.transferred), format_bytes(tot)) + } else { + format!("{}...", t!("downloading")) + }; + let btn_pause = Button::new(SharedString::from(format!("pause-{}", t.info.id))) + .ghost() + .small() + .icon(IconName::Pause) + .on_click(window.listener_for(&view, { + let id = t.info.id.clone(); + let tab_id = t.tab_id.clone(); + move |this, _, _, _| { + if let Some(handle) = this.sftp_handles.get(&tab_id) { + handle.pause_transfer(id.clone()); + } + } + })); + let btn_cancel = Button::new(SharedString::from(format!("cancel-{}", t.info.id))) + .ghost() + .small() + .icon(IconName::Close) + .on_click(window.listener_for(&view, { + let id = t.info.id.clone(); + let tab_id = t.tab_id.clone(); + move |this, _, _, _| { + if let Some(handle) = this.sftp_handles.get(&tab_id) { + handle.cancel_transfer(id.clone()); + } + } + })); + (txt, h_flex().gap_1().child(btn_pause).child(btn_cancel)) + } + crate::terminal::TransferState::Paused => { + let txt = t!("paused").to_string(); + let btn_resume = Button::new(SharedString::from(format!("resume-{}", t.info.id))) + .ghost() + .small() + .icon(IconName::Play) + .on_click(window.listener_for(&view, { + let id = t.info.id.clone(); + let tab_id = t.tab_id.clone(); + move |this, _, _, _| { + if let Some(handle) = this.sftp_handles.get(&tab_id) { + handle.resume_transfer(id.clone()); + } + } + })); + let btn_cancel = Button::new(SharedString::from(format!("cancel-{}", t.info.id))) + .ghost() + .small() + .icon(IconName::Close) + .on_click(window.listener_for(&view, { + let id = t.info.id.clone(); + let tab_id = t.tab_id.clone(); + move |this, _, _, _| { + if let Some(handle) = this.sftp_handles.get(&tab_id) { + handle.cancel_transfer(id.clone()); + } + } + })); + (txt, h_flex().gap_1().child(btn_resume).child(btn_cancel)) + } + crate::terminal::TransferState::Completed => { + let txt = t!("completed").to_string(); + let mut actions = h_flex().gap_1(); + if matches!(t.info.kind, crate::terminal::TransferType::Download) { + let btn_folder = Button::new(SharedString::from(format!("folder-{}", t.info.id))) + .ghost() + .small() + .icon(IconName::Folder) + .on_click({ + let target = t.info.target.clone(); + move |_, _, _| { + let _ = std::process::Command::new("open").arg(&target).spawn(); + } + }); + actions = actions.child(btn_folder); + } + (txt, actions) + } + crate::terminal::TransferState::Failed(ref err) => { + (format!("{}: {}", t!("failed"), err), h_flex().gap_1()) + } + crate::terminal::TransferState::Cancelled => { + (t!("cancelled").to_string(), h_flex().gap_1()) + } + }; + + let percent = match t.state { + crate::terminal::TransferState::Completed => 100.0, + _ => t.total.map(|tot| t.transferred as f64 / tot as f64 * 100.0).unwrap_or(0.0), + }; + + v_flex() + .gap_1() + .p_2() + .rounded_md() + .border_1() + .border_color(cx.theme().border) + .bg(cx.theme().muted) + .child( + h_flex() + .items_center() + .gap_2() + .child(Button::new(SharedString::from(format!("icon-{}", t.info.id))).icon(icon).ghost().small().disabled(true)) + .child( + v_flex() + .flex_1() + .min_w(px(0.)) + .overflow_hidden() + .child( + div() + .text_size(px(12.)) + .font_weight(FontWeight::SEMIBOLD) + .text_color(cx.theme().foreground) + .overflow_hidden() + .child(t.info.name.clone()), + ) + .child( + div() + .text_size(px(10.)) + .text_color(cx.theme().muted_foreground) + .overflow_hidden() + .child(format!("{}: {}", t!("session"), t.tab_title)), + ) + ) + .child( + div() + .text_size(px(11.)) + .text_color(cx.theme().muted_foreground) + .child(status_text), + ) + .child(actions), + ) + .when(matches!(t.state, crate::terminal::TransferState::Running | crate::terminal::TransferState::Paused), |this| { + this.child( + Progress::new(format!("progress-{}", t.info.id)) + .with_size(px(4.)) + .value(percent as f32) + .color(cx.theme().primary) + .w_full(), + ) + }) + })); + + let scroll_handle = window + .use_keyed_state("transfers-scroll", cx, |_, _| gpui::ScrollHandle::default()) + .read(cx) + .clone(); + + content.child( + div() + .w_full() + .relative() + .child( + div() + .w_full() + .max_h(px(400.)) + .flex_col() + .id("transfers-scroll-view") + .track_scroll(&scroll_handle) + .overflow_y_scroll() + .child(list) + ) + .child( + div() + .absolute() + .top_0() + .right_0() + .bottom_0() + .w(px(16.)) + .child( + Scrollbar::vertical(&scroll_handle) + .scrollbar_show(ScrollbarShow::Always) + ) + ) + ) + } + }) + }); + } + fn show_settings_dialog(&mut self, window: &mut Window, cx: &mut Context) { let view = cx.entity(); window.open_dialog(cx, move |dialog: Dialog, _window, _| { @@ -2534,26 +2798,114 @@ impl Ashell { _window: &mut Window, cx: &mut Context, ) -> impl IntoElement { - let Some(sftp) = self.active_sftp() else { - return v_flex() - .size_full() - .gap_3() - .p_3() - .border_color(cx.theme().border) - .bg(cx.theme().muted) - .child( - div() - .text_size(px(12.)) - .font_weight(FontWeight::SEMIBOLD) - .text_color(cx.theme().primary) - .child(t!("remote_files")), + let active_sftp = self.active_sftp(); + + let header = h_flex() + .flex_none() + .h(px(34.)) + .items_center() + .gap_2() + .px_3() + .border_b_1() + .border_color(cx.theme().border) + .bg(cx.theme().tab_bar) + .child( + div() + .text_size(px(12.)) + .font_weight(FontWeight::SEMIBOLD) + .text_color(cx.theme().primary) + .child(t!("remote_files")), + ) + .child(div().flex_1()) + .when_some(active_sftp.clone(), |this, sftp| { + let selected_entries = sftp.selected_entries.clone(); + this.child( + Button::new("sftp-refresh") + .ghost() + .small() + .icon(IconName::ArrowRight) + .label(t!("refresh").to_string()) + .on_click(cx.listener(|this, _, _, cx| this.refresh_sftp(cx))), ) .child( - div() - .text_size(px(12.)) - .text_color(cx.theme().muted_foreground) - .child(t!("open_ssh_tab_sftp")), - ); + Button::new("sftp-upload-file") + .ghost() + .small() + .icon(IconName::Plus) + .label(t!("upload_file").to_string()) + .on_click(cx.listener(|this, _, window, cx| { + this.upload_sftp_files(window, cx) + })), + ) + .child( + Button::new("sftp-upload-folder") + .ghost() + .small() + .icon(IconName::Folder) + .label(t!("upload_folder").to_string()) + .on_click(cx.listener(|this, _, window, cx| { + this.upload_sftp_folder(window, cx) + })), + ) + .child( + Button::new("sftp-download-selected") + .ghost() + .small() + .icon(IconName::ArrowDown) + .label(if selected_entries.is_empty() { + t!("download").to_string() + } else { + t!("download_count", count = selected_entries.len()).to_string() + }) + .disabled(selected_entries.is_empty()) + .on_click(cx.listener(|this, _, window, cx| { + this.download_selected_sftp_entries(window, cx); + })), + ) + .child( + Checkbox::new("sftp-show-hidden") + .small() + .label(t!("hidden").to_string()) + .checked(self.show_hidden_files) + .tab_stop(false) + .on_click(cx.listener(|this, checked, _, cx| { + this.show_hidden_files = *checked; + cx.notify(); + })), + ) + }) + .child( + Button::new("open-transfers") + .ghost() + .small() + .icon(IconName::ArrowDown) + .label(t!("transfers").to_string()) + .on_click(cx.listener(|this, _, window, cx| { + this.show_transfers_dialog(window, cx); + })), + ); + + let Some(sftp) = active_sftp else { + return v_flex() + .size_full() + .gap_0() + .border_color(cx.theme().border) + .bg(cx.theme().background) + .child(header) + .child( + v_flex() + .flex_1() + .items_center() + .justify_center() + .p_3() + .child( + div() + .text_size(px(12.)) + .text_color(cx.theme().muted_foreground) + .child(t!("open_ssh_tab_sftp")), + ), + ) + .into_any_element(); }; let selected_path = sftp.selected_path.clone(); @@ -2590,78 +2942,7 @@ impl Ashell { this.upload_sftp_files_batch(paths_to_upload, cx); }), ) - .child( - h_flex() - .h(px(34.)) - .items_center() - .gap_2() - .px_3() - .border_b_1() - .border_color(cx.theme().border) - .bg(cx.theme().tab_bar) - .child( - div() - .text_size(px(12.)) - .font_weight(FontWeight::SEMIBOLD) - .text_color(cx.theme().primary) - .child(t!("remote_files")), - ) - .child(div().flex_1()) - .child( - Button::new("sftp-refresh") - .ghost() - .small() - .icon(IconName::ArrowRight) - .label(t!("refresh").to_string()) - .on_click(cx.listener(|this, _, _, cx| this.refresh_sftp(cx))), - ) - .child( - Button::new("sftp-upload-file") - .ghost() - .small() - .icon(IconName::Plus) - .label(t!("upload_file").to_string()) - .on_click(cx.listener(|this, _, window, cx| { - this.upload_sftp_files(window, cx) - })), - ) - .child( - Button::new("sftp-upload-folder") - .ghost() - .small() - .icon(IconName::Folder) - .label(t!("upload_folder").to_string()) - .on_click(cx.listener(|this, _, window, cx| { - this.upload_sftp_folder(window, cx) - })), - ) - .child( - Button::new("sftp-download-selected") - .ghost() - .small() - .icon(IconName::ArrowDown) - .label(if selected_entries.is_empty() { - t!("download").to_string() - } else { - t!("download_count", count = selected_entries.len()).to_string() - }) - .disabled(selected_entries.is_empty()) - .on_click(cx.listener(|this, _, window, cx| { - this.download_selected_sftp_entries(window, cx); - })), - ) - .child( - Checkbox::new("sftp-show-hidden") - .small() - .label(t!("hidden").to_string()) - .checked(self.show_hidden_files) - .tab_stop(false) - .on_click(cx.listener(|this, checked, _, cx| { - this.show_hidden_files = *checked; - cx.notify(); - })), - ), - ) + .child(header) .child( h_flex() .h(px(36.)) @@ -2919,6 +3200,7 @@ impl Ashell { ) .child( h_flex() + .flex_none() .h(px(24.)) .px_3() .items_center() @@ -2934,6 +3216,7 @@ impl Ashell { .child(status), ), ) + .into_any_element() } fn sidebar(&self, cx: &mut Context) -> impl IntoElement { @@ -3186,6 +3469,10 @@ impl Render for Ashell { } self.sync_sftp_path_input(window, cx); self.sync_terminal_size(window, cx); + if self.show_transfers_dialog { + self.show_transfers_dialog = false; + self.show_transfers_dialog(window, cx); + } if let Some(new_display_offset) = self.terminal_scrollbar.future_display_offset.take() { if let Some(active_id) = self.active_tab.clone() { if let Some(tab) = self.tabs.iter_mut().find(|tab| tab.id == active_id) { @@ -3387,6 +3674,7 @@ impl Render for Ashell { ), ) .children(Root::render_dialog_layer(window, cx)) + .children(Root::render_sheet_layer(window, cx)) .when_some(self.sftp_context_menu.clone(), |this, menu| { let label = if menu.is_dir { diff --git a/src/sftp.rs b/src/sftp.rs index fa162ac..ca52347 100644 --- a/src/sftp.rs +++ b/src/sftp.rs @@ -62,9 +62,68 @@ pub enum SftpCommand { locals: Vec, remote_dir: String, }, + PauseTransfer(String), + ResumeTransfer(String), + CancelTransfer(String), + TransferFinished(String), Close, } +use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; + +pub struct TransferStateFlag(pub Arc); + +impl TransferStateFlag { + pub fn new() -> Self { + Self(Arc::new(AtomicU8::new(0))) + } + + pub fn pause(&self) { self.0.store(1, Ordering::SeqCst); } + pub fn resume(&self) { self.0.store(0, Ordering::SeqCst); } + pub fn cancel(&self) { self.0.store(2, Ordering::SeqCst); } + + pub async fn yield_if_paused( + &self, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, + transferred: u64, + total: Option, + ) -> anyhow::Result<()> { + let mut was_paused = false; + loop { + let state = self.0.load(Ordering::SeqCst); + if state == 2 { + return Err(anyhow::anyhow!("transfer cancelled")); + } + if state == 1 { + if !was_paused { + let _ = events.send(crate::terminal::BackendEvent::TransferProgress { + tab_id: tab_id.to_string(), + id: id.to_string(), + transferred, + total, + state: crate::terminal::TransferState::Paused, + }); + was_paused = true; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } else { + if was_paused { + let _ = events.send(crate::terminal::BackendEvent::TransferProgress { + tab_id: tab_id.to_string(), + id: id.to_string(), + transferred, + total, + state: crate::terminal::TransferState::Running, + }); + } + return Ok(()); + } + } + } +} + pub struct SftpHandle { pub commands: UnboundedSender, #[allow(dead_code)] @@ -105,6 +164,18 @@ impl SftpHandle { pub fn close(&self) { let _ = self.commands.send(SftpCommand::Close); } + + pub fn pause_transfer(&self, id: String) { + let _ = self.commands.send(SftpCommand::PauseTransfer(id)); + } + + pub fn resume_transfer(&self, id: String) { + let _ = self.commands.send(SftpCommand::ResumeTransfer(id)); + } + + pub fn cancel_transfer(&self, id: String) { + let _ = self.commands.send(SftpCommand::CancelTransfer(id)); + } } pub fn spawn_sftp( @@ -114,8 +185,9 @@ pub fn spawn_sftp( events: std::sync::mpsc::Sender, ) -> SftpHandle { let (cmd_tx, cmd_rx) = mpsc::unbounded_channel(); + let cmd_tx_clone = cmd_tx.clone(); let join = runtime.spawn(async move { - if let Err(err) = run_sftp(tab_id.clone(), session, cmd_rx, events.clone()).await { + if let Err(err) = run_sftp(tab_id.clone(), session, cmd_rx, cmd_tx_clone, events.clone()).await { let _ = events.send(BackendEvent::SftpStatus { tab_id, text: format!("sftp error: {err:#}"), @@ -132,11 +204,12 @@ async fn run_sftp( tab_id: String, session: Session, mut commands: UnboundedReceiver, + commands_tx: UnboundedSender, events: std::sync::mpsc::Sender, ) -> Result<()> { let _ = events.send(BackendEvent::SftpStatus { tab_id: tab_id.clone(), - text: "sftp connecting...".into(), + text: t!("sftp_connecting").to_string(), }); let handle = connect_and_authenticate(&session).await?; @@ -158,9 +231,29 @@ async fn run_sftp( .unwrap_or_else(|_| "/".to_string()); emit_entries(&events, &tab_id, &sftp, &home).await?; + let mut active_transfers: std::collections::HashMap = std::collections::HashMap::new(); + while let Some(command) = commands.recv().await { match command { SftpCommand::Close => break, + SftpCommand::PauseTransfer(id) => { + if let Some(flag) = active_transfers.get(&id) { + flag.pause(); + } + } + SftpCommand::ResumeTransfer(id) => { + if let Some(flag) = active_transfers.get(&id) { + flag.resume(); + } + } + SftpCommand::CancelTransfer(id) => { + if let Some(flag) = active_transfers.remove(&id) { + flag.cancel(); + } + } + SftpCommand::TransferFinished(id) => { + active_transfers.remove(&id); + } SftpCommand::ListDir(path) => { let actual_path = if path == "~" { home.clone() @@ -192,46 +285,142 @@ async fn run_sftp( } }, SftpCommand::Download { remote, local_dir } => { - let base = base_name(&remote); - let _ = events.send(BackendEvent::SftpStatus { + let id = uuid::Uuid::new_v4().to_string(); + let flag = TransferStateFlag::new(); + active_transfers.insert(id.clone(), TransferStateFlag(flag.0.clone())); + + let info = crate::terminal::TransferInfo { + id: id.clone(), + name: base_name(&remote).to_string(), + source: remote.clone(), + target: local_dir.clone(), + kind: crate::terminal::TransferType::Download, + total_bytes: None, + }; + let _ = events.send(BackendEvent::TransferStarted { tab_id: tab_id.clone(), - text: t!("downloading_file", base = base).into(), + info, }); - match download_path_impl(&handle, &sftp, &remote, Path::new(&local_dir)).await { - Ok(summary) => { - let _ = events.send(BackendEvent::SftpStatus { - tab_id: tab_id.clone(), - text: summary.into(), - }); + + let handle_clone = handle.clone(); + let events_clone = events.clone(); + let tab_id_clone = tab_id.clone(); + let commands_tx_clone = commands_tx.clone(); + + tokio::spawn(async move { + let Ok(channel) = handle_clone.channel_open_session().await else { return }; + let Ok(_) = channel.request_subsystem(true, "sftp").await else { return }; + let Ok(sftp_session) = SftpSession::new(channel.into_stream()).await else { return }; + + let _ = events_clone.send(BackendEvent::SftpStatus { + tab_id: tab_id_clone.clone(), + text: t!("downloading_file", base = base_name(&remote)).to_string(), + }); + + match download_path_impl(&handle_clone, &sftp_session, &remote, Path::new(&local_dir), flag, &events_clone, &tab_id_clone, &id).await { + Ok(summary) => { + let _ = events_clone.send(BackendEvent::SftpStatus { + tab_id: tab_id_clone, + text: summary, + }); + } + Err(err) => { + let err_msg = format!("{err:#}"); + let _ = events_clone.send(BackendEvent::SftpStatus { + tab_id: tab_id_clone.clone(), + text: t!("download_failed", err = err_msg.clone()).to_string(), + }); + let _ = events_clone.send(BackendEvent::TransferProgress { + tab_id: tab_id_clone, + id: id.clone(), + transferred: 0, + total: None, + state: crate::terminal::TransferState::Failed(err_msg), + }); + } } - Err(err) => { - let _ = events.send(BackendEvent::SftpStatus { - tab_id: tab_id.clone(), - text: t!("download_failed", err = format!("{err:#}")).into(), - }); - } - } + let _ = commands_tx_clone.send(SftpCommand::TransferFinished(id)); + }); } SftpCommand::UploadPaths { locals, remote_dir } => { - let _ = events.send(BackendEvent::SftpStatus { + let id = uuid::Uuid::new_v4().to_string(); + let flag = TransferStateFlag::new(); + active_transfers.insert(id.clone(), TransferStateFlag(flag.0.clone())); + + let name = if locals.len() == 1 { + base_name(&locals[0]).to_string() + } else { + let mut file_count = 0; + let mut folder_count = 0; + for local in &locals { + if std::path::Path::new(local).is_dir() { + folder_count += 1; + } else { + file_count += 1; + } + } + if file_count > 0 && folder_count == 0 { + t!("n_files", files = file_count).to_string() + } else if file_count == 0 && folder_count > 0 { + t!("n_folders", folders = folder_count).to_string() + } else { + t!("n_files_and_folders", files = file_count, folders = folder_count).to_string() + } + }; + + let info = crate::terminal::TransferInfo { + id: id.clone(), + name, + source: "local".to_string(), + target: remote_dir.clone(), + kind: crate::terminal::TransferType::Upload, + total_bytes: None, + }; + let _ = events.send(BackendEvent::TransferStarted { tab_id: tab_id.clone(), - text: t!("uploading").into(), + info, }); - match upload_paths_impl(&sftp, &locals, &remote_dir).await { - Ok(summary) => { - let _ = events.send(BackendEvent::SftpStatus { - tab_id: tab_id.clone(), - text: summary.into(), - }); - let _ = emit_entries(&events, &tab_id, &sftp, &remote_dir).await; + + let handle_clone = handle.clone(); + let events_clone = events.clone(); + let tab_id_clone = tab_id.clone(); + let commands_tx_clone = commands_tx.clone(); + + tokio::spawn(async move { + let Ok(channel) = handle_clone.channel_open_session().await else { return }; + let Ok(_) = channel.request_subsystem(true, "sftp").await else { return }; + let Ok(sftp_session) = SftpSession::new(channel.into_stream()).await else { return }; + + let _ = events_clone.send(BackendEvent::SftpStatus { + tab_id: tab_id_clone.clone(), + text: t!("uploading").to_string(), + }); + + match upload_paths_impl(&sftp_session, &locals, &remote_dir, flag, &events_clone, &tab_id_clone, &id).await { + Ok(summary) => { + let _ = events_clone.send(BackendEvent::SftpStatus { + tab_id: tab_id_clone.clone(), + text: summary, + }); + let _ = commands_tx_clone.send(SftpCommand::ListDir(remote_dir)); + } + Err(err) => { + let err_msg = format!("{err:#}"); + let _ = events_clone.send(BackendEvent::SftpStatus { + tab_id: tab_id_clone.clone(), + text: t!("upload_failed", err = err_msg.clone()).to_string(), + }); + let _ = events_clone.send(BackendEvent::TransferProgress { + tab_id: tab_id_clone, + id: id.clone(), + transferred: 0, + total: None, + state: crate::terminal::TransferState::Failed(err_msg), + }); + } } - Err(err) => { - let _ = events.send(BackendEvent::SftpStatus { - tab_id: tab_id.clone(), - text: t!("upload_failed", err = format!("{err:#}")).into(), - }); - } - } + let _ = commands_tx_clone.send(SftpCommand::TransferFinished(id)); + }); } } } @@ -263,7 +452,7 @@ async fn emit_entries( async fn connect_and_authenticate( session: &Session, -) -> Result> { +) -> Result>> { let config = Arc::new(client::Config { inactivity_timeout: Some(std::time::Duration::from_secs(600)), ..Default::default() @@ -304,7 +493,7 @@ async fn connect_and_authenticate( )); } - Ok(handle) + Ok(Arc::new(handle)) } fn load_session_private_key(session: &Session) -> Result { @@ -523,6 +712,10 @@ async fn download_path_impl( sftp: &SftpSession, remote: &str, local_dir: &Path, + flag: TransferStateFlag, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, ) -> Result { tokio::fs::create_dir_all(local_dir) .await @@ -544,12 +737,12 @@ async fn download_path_impl( Uuid::new_v4() )); let extracted_to = - download_remote_directory_archive(handle, sftp, remote, &local_archive).await?; + download_remote_directory_archive(handle, sftp, remote, &local_archive, &flag, events, tab_id, id).await?; return Ok(t!("downloaded_folder", path = extracted_to.display()).to_string()); } let local_path = local_dir.join(base_name(remote)); - download_file_impl(sftp, remote, &local_path).await?; + download_file_impl(sftp, remote, &local_path, &flag, events, tab_id, id).await?; Ok(t!("downloaded_file", path = local_path.display()).to_string()) } @@ -558,6 +751,10 @@ async fn download_dir_recursive( sftp: &SftpSession, remote_dir: &str, local_dir: &Path, + flag: &TransferStateFlag, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, ) -> Result<()> { tokio::fs::create_dir_all(local_dir) .await @@ -566,9 +763,9 @@ async fn download_dir_recursive( for entry in entries { let local_path = local_dir.join(&entry.name); if entry.is_dir { - Box::pin(download_dir_recursive(sftp, &entry.full_path, &local_path)).await?; + Box::pin(download_dir_recursive(sftp, &entry.full_path, &local_path, flag, events, tab_id, id)).await?; } else { - download_file_impl(sftp, &entry.full_path, &local_path).await?; + download_file_impl(sftp, &entry.full_path, &local_path, flag, events, tab_id, id).await?; let _ = maybe_extract_archive(&local_path).await; } } @@ -580,6 +777,10 @@ async fn download_remote_directory_archive( sftp: &SftpSession, remote_dir: &str, local_archive: &Path, + flag: &TransferStateFlag, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, ) -> Result { let remote_archive = format!( "/tmp/ashell-{}-{}.tar.gz", @@ -593,7 +794,7 @@ async fn download_remote_directory_archive( .join(base_name(remote_dir)); let archive_download = async { - download_file_impl(sftp, &remote_archive, local_archive).await?; + download_file_impl(sftp, &remote_archive, local_archive, flag, events, tab_id, id).await?; extract_archive_to( local_archive, local_archive.parent().unwrap_or_else(|| Path::new(".")), @@ -616,7 +817,15 @@ async fn download_remote_directory_archive( Ok(extracted_to) } -async fn download_file_impl(sftp: &SftpSession, remote: &str, local: &Path) -> Result<()> { +async fn download_file_impl( + sftp: &SftpSession, + remote: &str, + local: &Path, + flag: &TransferStateFlag, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, +) -> Result<()> { let mut remote_file = sftp .open(remote) .await @@ -625,8 +834,12 @@ async fn download_file_impl(sftp: &SftpSession, remote: &str, local: &Path) -> R .await .with_context(|| format!("create local {}", local.display()))?; - let mut buffer = vec![0u8; 64 * 1024]; + let total = sftp.metadata(remote).await.ok().and_then(|m| m.size); + let mut transferred = 0u64; + + let mut buffer = vec![0u8; 128 * 1024]; loop { + flag.yield_if_paused(events, tab_id, id, transferred, total).await?; let read = remote_file .read(&mut buffer) .await @@ -638,8 +851,26 @@ async fn download_file_impl(sftp: &SftpSession, remote: &str, local: &Path) -> R .write_all(&buffer[..read]) .await .with_context(|| format!("write {}", local.display()))?; + + transferred += read as u64; + let _ = events.send(BackendEvent::TransferProgress { + tab_id: tab_id.to_string(), + id: id.to_string(), + transferred, + total, + state: crate::terminal::TransferState::Running, + }); } local_file.flush().await.context("flush local file")?; + + let _ = events.send(BackendEvent::TransferProgress { + tab_id: tab_id.to_string(), + id: id.to_string(), + transferred, + total, + state: crate::terminal::TransferState::Completed, + }); + Ok(()) } @@ -647,92 +878,126 @@ async fn upload_paths_impl( sftp: &SftpSession, locals: &[String], remote_dir: &str, + flag: TransferStateFlag, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, ) -> Result { create_remote_dir_all(sftp, remote_dir).await?; let mut file_count = 0usize; let mut folder_count = 0usize; + + let mut total_bytes = 0u64; + let mut files_to_upload = Vec::new(); + let mut dirs_to_create = Vec::new(); + for local in locals { - let path = PathBuf::from(local); - if path.is_dir() { - upload_directory_impl(sftp, &path, remote_dir).await?; + let p = PathBuf::from(local); + if p.is_dir() { folder_count += 1; - } else { - upload_file_to_dir_impl(sftp, &path, remote_dir).await?; + let root_name = p.file_name().and_then(|n| n.to_str()).unwrap_or("folder"); + let remote_root = join_remote(remote_dir, root_name); + dirs_to_create.push(remote_root.clone()); + + for entry in WalkDir::new(&p) { + let entry = entry?; + let path = entry.path(); + if path == p { continue; } + + if let Ok(meta) = tokio::fs::metadata(&path).await { + let relative = path.strip_prefix(&p)?; + let remote_path = if relative.as_os_str().is_empty() { + remote_root.clone() + } else { + let rel = relative.components().map(|c| c.as_os_str().to_string_lossy().to_string()).collect::>().join("/"); + join_remote(&remote_root, &rel) + }; + + if path.is_dir() { + dirs_to_create.push(remote_path); + } else { + total_bytes += meta.len(); + files_to_upload.push((path.to_path_buf(), remote_path)); + } + } + } + } else if let Ok(meta) = tokio::fs::metadata(&p).await { + total_bytes += meta.len(); + let file_name = p.file_name().and_then(|n| n.to_str()).unwrap_or("file"); + files_to_upload.push((p.clone(), join_remote(remote_dir, file_name))); file_count += 1; } } - let summary = match (file_count, folder_count) { - (1, 0) => t!("uploaded_file").to_string(), - (0, 1) => t!("uploaded_folder").to_string(), - (files, 0) => t!("uploaded_n_files", files = files).to_string(), - (0, folders) => t!("uploaded_n_folders", folders = folders).to_string(), - (files, folders) => t!( - "uploaded_files_and_folders", - files = files, - folders = folders - ) - .to_string(), + + // Create directories sequentially first + for dir in dirs_to_create { + create_remote_dir_all(sftp, &dir).await?; + } + + let transferred = Arc::new(AtomicU64::new(0)); + let mut futures = Vec::new(); + + for (local_path, remote_path) in files_to_upload { + let flag_clone = TransferStateFlag(Arc::clone(&flag.0)); + let events_clone = events.clone(); + let tab_id_clone = tab_id.to_string(); + let id_clone = id.to_string(); + let transferred_clone = Arc::clone(&transferred); + + futures.push(async move { + upload_file_impl( + sftp, + &local_path, + &remote_path, + &flag_clone, + &events_clone, + &tab_id_clone, + &id_clone, + transferred_clone, + Some(total_bytes), + ).await + }); + } + + use futures::StreamExt as _; + let mut stream = futures::stream::iter(futures).buffer_unordered(4); + while let Some(res) = stream.next().await { + res?; + } + + let _ = events.send(BackendEvent::TransferProgress { + tab_id: tab_id.to_string(), + id: id.to_string(), + transferred: total_bytes, + total: Some(total_bytes), + state: crate::terminal::TransferState::Completed, + }); + + let summary = if file_count == 1 && folder_count == 0 { + t!("uploaded_file").to_string() + } else if file_count == 0 && folder_count == 1 { + t!("uploaded_folder").to_string() + } else if file_count > 0 && folder_count == 0 { + t!("uploaded_n_files", files = file_count).to_string() + } else if file_count == 0 && folder_count > 0 { + t!("uploaded_n_folders", folders = folder_count).to_string() + } else { + t!("uploaded_files_and_folders", files = file_count, folders = folder_count).to_string() }; Ok(summary) } -async fn upload_directory_impl( - sftp: &SftpSession, - local_dir: &Path, - remote_parent: &str, +async fn upload_file_impl( + sftp: &SftpSession, + local_file: &Path, + remote_path: &str, + flag: &TransferStateFlag, + events: &std::sync::mpsc::Sender, + tab_id: &str, + id: &str, + transferred: Arc, + total: Option, ) -> Result<()> { - let root_name = local_dir - .file_name() - .and_then(|name| name.to_str()) - .ok_or_else(|| anyhow!("invalid folder name: {}", local_dir.display()))?; - let remote_root = join_remote(remote_parent, root_name); - create_remote_dir_all(sftp, &remote_root).await?; - - for entry in WalkDir::new(local_dir) { - let entry = entry?; - let path = entry.path(); - if path == local_dir { - continue; - } - let relative = path.strip_prefix(local_dir)?; - let remote_path = if relative.as_os_str().is_empty() { - remote_root.clone() - } else { - let rel = relative - .components() - .map(|component| component.as_os_str().to_string_lossy().to_string()) - .collect::>() - .join("/"); - join_remote(&remote_root, &rel) - }; - if path.is_dir() { - create_remote_dir_all(sftp, &remote_path).await?; - } else { - if let Some(parent) = Path::new(&remote_path).parent() { - let parent_remote = parent.to_string_lossy().replace('\\', "/"); - create_remote_dir_all(sftp, &parent_remote).await?; - } - upload_file_impl(sftp, path, &remote_path).await?; - } - } - - Ok(()) -} - -async fn upload_file_to_dir_impl( - sftp: &SftpSession, - local_file: &Path, - remote_dir: &str, -) -> Result<()> { - let file_name = local_file - .file_name() - .and_then(|name| name.to_str()) - .ok_or_else(|| anyhow!("invalid file name: {}", local_file.display()))?; - let remote_path = join_remote(remote_dir, file_name); - upload_file_impl(sftp, local_file, &remote_path).await -} - -async fn upload_file_impl(sftp: &SftpSession, local_file: &Path, remote_path: &str) -> Result<()> { let mut local = tokio::fs::File::open(local_file) .await .with_context(|| format!("open local {}", local_file.display()))?; @@ -741,8 +1006,10 @@ async fn upload_file_impl(sftp: &SftpSession, local_file: &Path, remote_path: &s .await .with_context(|| format!("create remote {remote_path}"))?; - let mut buffer = vec![0u8; 64 * 1024]; + let mut buffer = vec![0u8; 128 * 1024]; loop { + let cur = transferred.load(Ordering::Relaxed); + flag.yield_if_paused(events, tab_id, id, cur, total).await?; let read = local.read(&mut buffer).await.context("read local file")?; if read == 0 { break; @@ -751,6 +1018,15 @@ async fn upload_file_impl(sftp: &SftpSession, local_file: &Path, remote_path: &s .write_all(&buffer[..read]) .await .with_context(|| format!("write remote {remote_path}"))?; + + let new_cur = transferred.fetch_add(read as u64, Ordering::Relaxed) + read as u64; + let _ = events.send(BackendEvent::TransferProgress { + tab_id: tab_id.to_string(), + id: id.to_string(), + transferred: new_cur, + total, + state: crate::terminal::TransferState::Running, + }); } remote.flush().await.context("flush remote file")?; Ok(()) @@ -987,6 +1263,7 @@ async fn extract_archive_to(path: &Path, target_dir: &Path) -> Result<()> { Ok(()) } +#[derive(Clone)] struct SftpClientHandler; #[async_trait] diff --git a/src/terminal.rs b/src/terminal.rs index 24948d0..c4493f5 100644 --- a/src/terminal.rs +++ b/src/terminal.rs @@ -61,6 +61,18 @@ pub enum BackendEvent { tab_id: String, reason: String, }, + TransferProgress { + #[allow(dead_code)] + tab_id: String, + id: String, + transferred: u64, + total: Option, + state: TransferState, + }, + TransferStarted { + tab_id: String, + info: TransferInfo, + }, Closed { tab_id: String, reason: String, @@ -187,7 +199,7 @@ impl TerminalTab { tab.connected = false; tab.sftp = Some(SftpUiState { current_path: "/".into(), - status: "sftp connecting...".into(), + status: t!("sftp_connecting").to_string(), entries: Vec::new(), selected_path: None, preview: None, @@ -425,6 +437,7 @@ impl EventListener for TerminalListener { } } +use rust_i18n::t; fn new_term( cols: u16, rows: u16, @@ -648,3 +661,38 @@ fn modifier_code(keystroke: &Keystroke) -> u32 { } modifier_code + 1 } + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub enum TransferType { + Upload, + Download, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub enum TransferState { + Running, + Paused, + Completed, + Failed(String), + Cancelled, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct TransferInfo { + pub id: String, + pub name: String, + pub source: String, + pub target: String, + pub kind: TransferType, + pub total_bytes: Option, +} + +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct Transfer { + pub tab_id: String, + pub tab_title: String, + pub info: TransferInfo, + pub transferred: u64, + pub total: Option, + pub state: TransferState, +}