From 601cb10b32f9458c966451549d9c0fc1ce010153 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=89=E5=92=B2=E9=9B=85=20misaki=20masa?= Date: Fri, 29 May 2026 15:33:03 +0800 Subject: [PATCH] feat: Unix domain socket on Windows as the underlying transport layer for DDS (#4009) --- Cargo.lock | 31 +++++++++++++++++ yazi-dds/Cargo.toml | 3 ++ yazi-dds/src/client.rs | 4 +-- yazi-dds/src/state.rs | 3 +- yazi-dds/src/stream.rs | 76 ++++++++++++++++++++++++++++++------------ 5 files changed, 92 insertions(+), 25 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e2c3f2dc..7478fae3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1438,6 +1438,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd2e7510819d6fbf51a5545c8f922716ecfb14df168a3242f7d33e0239efe6a1" +[[package]] +name = "fastrand" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" + [[package]] name = "fax" version = "0.2.7" @@ -4319,6 +4325,19 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.4.2", + "once_cell", + "rustix 1.1.4", + "windows-sys 0.61.2", +] + [[package]] name = "terminfo" version = "0.9.0" @@ -4741,6 +4760,17 @@ version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" +[[package]] +name = "uds_windows" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e" +dependencies = [ + "memoffset", + "tempfile", + "windows-sys 0.61.2", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -5834,6 +5864,7 @@ dependencies = [ "tokio", "tokio-stream", "tracing", + "uds_windows", "vergen-gitcl", "yazi-binding", "yazi-boot", diff --git a/yazi-dds/Cargo.toml b/yazi-dds/Cargo.toml index f8f50c2f..cd4a81e5 100644 --- a/yazi-dds/Cargo.toml +++ b/yazi-dds/Cargo.toml @@ -40,3 +40,6 @@ tracing = { workspace = true } [build-dependencies] vergen-gitcl = { version = "9.1.0", features = [ "build" ] } + +[target.'cfg(windows)'.dependencies] +uds_windows = "1.2.1" diff --git a/yazi-dds/src/client.rs b/yazi-dds/src/client.rs index e193e215..919af9fd 100644 --- a/yazi-dds/src/client.rs +++ b/yazi-dds/src/client.rs @@ -48,8 +48,8 @@ impl Client { writer.write_all(payload.as_bytes()).await.ok(); // Retry once } } - Ok(next) = lines.next_line() => { - let Some(line) = next else { + next = lines.next_line() => { + let Ok(Some(line)) = next else { (lines, writer) = Self::reconnect(&mut server).await; continue; }; diff --git a/yazi-dds/src/state.rs b/yazi-dds/src/state.rs index 830f9901..c740921a 100644 --- a/yazi-dds/src/state.rs +++ b/yazi-dds/src/state.rs @@ -26,6 +26,7 @@ impl Deref for State { impl State { pub fn set(&self, kind: &str, sender: u64, body: &str) -> bool { + debug_assert!(kind.starts_with('@')); let Some(inner) = &mut *self.inner.write() else { return false }; if body == "null" { @@ -84,7 +85,7 @@ impl State { let line = mem::take(&mut buf); let mut parts = line.splitn(4, ','); - let Some(kind) = parts.next() else { continue }; + let Some(kind) = parts.next().filter(|s| s.starts_with('@')) else { continue }; let Some(_) = parts.next() else { continue }; inner.insert(kind.to_owned(), line); } diff --git a/yazi-dds/src/stream.rs b/yazi-dds/src/stream.rs index b09a6a0f..5aba44bd 100644 --- a/yazi-dds/src/stream.rs +++ b/yazi-dds/src/stream.rs @@ -1,25 +1,24 @@ -use std::io; +use std::{io, path::PathBuf}; -use tokio::io::{BufReader, Lines, ReadHalf, WriteHalf}; +use tokio::{io::{AsyncBufReadExt, BufReader, Lines, ReadHalf, WriteHalf}, sync::OnceCell}; +use yazi_fs::{Xdg, create_owned_dir, provider::{Provider, local::Local}}; pub struct Stream; -use tokio::io::AsyncBufReadExt; - #[cfg(unix)] pub type ClientReader = Lines>>; -#[cfg(not(unix))] +#[cfg(windows)] pub type ClientReader = Lines>>; #[cfg(unix)] pub(super) type ClientWriter = WriteHalf; -#[cfg(not(unix))] +#[cfg(windows)] pub(super) type ClientWriter = WriteHalf; #[cfg(unix)] pub(super) type ServerListener = tokio::net::UnixListener; -#[cfg(not(unix))] -pub(super) type ServerListener = tokio::net::TcpListener; +#[cfg(windows)] +pub(super) type ServerListener = WinUnixListener; impl Stream { #[cfg(unix)] @@ -29,34 +28,36 @@ impl Stream { Ok((BufReader::new(reader).lines(), writer)) } - #[cfg(not(unix))] + #[cfg(windows)] pub async fn connect() -> io::Result<(ClientReader, ClientWriter)> { - let stream = tokio::net::TcpStream::connect("127.0.0.1:33581").await?; - let (reader, writer) = tokio::io::split(stream); + let p = Self::socket_file().await?; + let uds = tokio::task::spawn_blocking(move || uds_windows::UnixStream::connect(p)).await??; + + let (reader, writer) = tokio::io::split(WinUnixListener::into_tokio(uds)?); Ok((BufReader::new(reader).lines(), writer)) } #[cfg(unix)] pub(super) async fn bind() -> io::Result { - use yazi_fs::provider::Provider; - let p = Self::socket_file().await?; - yazi_fs::provider::local::Local::regular(&p).remove_file().await.ok(); + Local::regular(&p).remove_file().await.ok(); tokio::net::UnixListener::bind(p) } - #[cfg(not(unix))] + #[cfg(windows)] pub(super) async fn bind() -> io::Result { - tokio::net::TcpListener::bind("127.0.0.1:33581").await + let p = Self::socket_file().await?; + Local::regular(&p).remove_file().await.ok(); + + let listener = uds_windows::UnixListener::bind(p)?; + listener.set_nonblocking(true)?; + + Ok(WinUnixListener(listener)) } - #[cfg(unix)] - async fn socket_file() -> io::Result<&'static std::path::PathBuf> { - use tokio::sync::OnceCell; - use yazi_fs::{Xdg, create_owned_dir}; - - static ONCE: OnceCell = OnceCell::const_new(); + async fn socket_file() -> io::Result<&'static PathBuf> { + static ONCE: OnceCell = OnceCell::const_new(); ONCE .get_or_try_init(|| async move { let p = Xdg::runtime_dir(); @@ -67,3 +68,34 @@ impl Stream { .await } } + +// --- WinUnixListener +#[cfg(windows)] +pub(super) struct WinUnixListener(uds_windows::UnixListener); + +#[cfg(windows)] +impl WinUnixListener { + pub(super) async fn accept( + &self, + ) -> io::Result<(tokio::net::TcpStream, uds_windows::SocketAddr)> { + loop { + match self.0.accept() { + Ok((stream, addr)) => return Ok((Self::into_tokio(stream)?, addr)), + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + Err(e) => return Err(e), + } + } + } + + fn into_tokio(uds: uds_windows::UnixStream) -> io::Result { + use std::os::windows::io::{FromRawSocket, IntoRawSocket}; + + let raw = uds.into_raw_socket(); + let std = unsafe { std::net::TcpStream::from_raw_socket(raw) }; + std.set_nonblocking(true)?; + + tokio::net::TcpStream::from_std(std) + } +}