starred/ashell
1
0
Fork 0
mirror of https://github.com/rust-kotlin/ashell.git synced 2026-09-22 00:00:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor SFTP transfers with concurrency, persistence, and UI improvements

Browse Source 
- 使用多线程并发处理多文件/大文件 SFTP 传输以提升速度
- 修复 SFTP 底部状态栏不显示的排版问题
- 新增持久化功能：将传输历史记录保存至本地配置文件夹并在启动时加载
- 修复传输记录中文件和文件夹数量统计及 i18n 汉化的正确显示
- 优化传输记录窗口 UI：增加自适应滚动列表，并集成匹配 SFTP 面板原生样式的右侧长驻滚动条
- 清理 Cargo warning，修复无用变量等问题
This commit is contained in:
TomZz
2026-06-12 00:26:04 +08:00
parent 7504bf6448
commit 06796031e6
8 changed files with 860 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -292,6 +292,7 @@ dependencies = [
"chrono",
"directories",
"flate2",
"futures",
"gpui",
"gpui-component",
"gpui-component-assets",
Cargo.toml
+1
View File
@@ -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
locales/en.yml
+12
View File
@@ -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"
locales/zh-CN.yml
+12
View File
@@ -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: "会话"
src/config.rs
+13
View File
@@ -115,6 +115,8 @@ pub struct ConfigFile {
pub workspace_panels: Option<Vec<f32>>,
#[serde(default)]
pub body_panels: Option<Vec<f32>>,
#[serde(default)]
pub transfers: Vec<crate::terminal::Transfer>,
}
fn default_locale() -> String {
@@ -230,6 +232,17 @@ impl ConfigStore {
self.cache.body_panels.as_ref()
}
pub fn transfers(&self) -> Vec<crate::terminal::Transfer> {
self.cache.transfers.clone()
}
pub fn set_transfers(&mut self, transfers: Vec<crate::terminal::Transfer>) {
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<SavedWindowBounds>,
src/main.rs
+379 -91
View File
@@ -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<String>,
sftp_context_menu: Option<SftpContextMenuState>,
show_hidden_files: bool,
transfers: Vec<crate::terminal::Transfer>,
show_transfers_dialog: bool,
system_status: Option<SharedString>,
terminal_bounds: Option<Bounds<Pixels>>,
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<Self>) {
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<Self>) {
let view = cx.entity();
window.open_dialog(cx, move |dialog: Dialog, _window, _| {
@@ -2534,26 +2798,114 @@ impl Ashell {
_window: &mut Window,
cx: &mut Context<Self>,
) -> 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<Self>) -> 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 {
src/sftp.rs
+393 -116
View File
@@ -62,9 +62,68 @@ pub enum SftpCommand {
locals: Vec<String>,
remote_dir: String,
},
PauseTransfer(String),
ResumeTransfer(String),
CancelTransfer(String),
TransferFinished(String),
Close,
}
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
pub struct TransferStateFlag(pub Arc<AtomicU8>);
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<crate::terminal::BackendEvent>,
tab_id: &str,
id: &str,
transferred: u64,
total: Option<u64>,
) -> 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<SftpCommand>,
#[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<BackendEvent>,
) -> 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<SftpCommand>,
commands_tx: UnboundedSender<SftpCommand>,
events: std::sync::mpsc::Sender<BackendEvent>,
) -> 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<String, TransferStateFlag> = 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<russh::client::Handle<SftpClientHandler>> {
) -> Result<Arc<russh::client::Handle<SftpClientHandler>>> {
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<PrivateKey> {
@@ -523,6 +712,10 @@ async fn download_path_impl(
sftp: &SftpSession,
remote: &str,
local_dir: &Path,
flag: TransferStateFlag,
events: &std::sync::mpsc::Sender<BackendEvent>,
tab_id: &str,
id: &str,
) -> Result<String> {
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<BackendEvent>,
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<BackendEvent>,
tab_id: &str,
id: &str,
) -> Result<PathBuf> {
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<BackendEvent>,
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<BackendEvent>,
tab_id: &str,
id: &str,
) -> Result<String> {
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::<Vec<_>>().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<BackendEvent>,
tab_id: &str,
id: &str,
transferred: Arc<AtomicU64>,
total: Option<u64>,
) -> 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::<Vec<_>>()
.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]
src/terminal.rs
+49 -1
View File
@@ -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<u64>,
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<u64>,
}
#[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<u64>,
pub state: TransferState,
}