feat: Unix domain socket on Windows as the underlying transport layer for DDS (#4009)

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",

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"

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;
 						};

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);
 		}

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<BufReader<ReadHalf<tokio::net::UnixStream>>>;
-#[cfg(not(unix))]
+#[cfg(windows)]
 pub type ClientReader = Lines<BufReader<ReadHalf<tokio::net::TcpStream>>>;
 
 #[cfg(unix)]
 pub(super) type ClientWriter = WriteHalf<tokio::net::UnixStream>;
-#[cfg(not(unix))]
+#[cfg(windows)]
 pub(super) type ClientWriter = WriteHalf<tokio::net::TcpStream>;
 
 #[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<ServerListener> {
-		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<ServerListener> {
-		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<std::path::PathBuf> = OnceCell::const_new();
+	async fn socket_file() -> io::Result<&'static PathBuf> {
+		static ONCE: OnceCell<PathBuf> = 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<tokio::net::TcpStream> {
+		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)
+	}
+}