use std::io::{BufRead, BufReader, Write}; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; use anyhow::{anyhow, bail, Result}; use serde_json::{json, Value}; /// Platform-specific transport for mpv's JSON IPC. /// Unix: a Unix domain socket. Windows: a named pipe (`\\.\pipe\...`). #[cfg(unix)] mod ipc { use std::io; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; pub type Reader = UnixStream; pub type Writer = UnixStream; pub fn endpoint() -> PathBuf { let dir = dirs::runtime_dir().unwrap_or_else(std::env::temp_dir); let path = dir.join(format!("drift-{}.sock", std::process::id())); let _ = std::fs::remove_file(&path); path } pub fn connect(path: &Path) -> io::Result<(Reader, Writer)> { let stream = UnixStream::connect(path)?; let reader = stream.try_clone()?; Ok((reader, stream)) } pub fn cleanup(path: &Path) { let _ = std::fs::remove_file(path); } } #[cfg(windows)] mod ipc { use std::ffi::c_void; use std::fs::{File, OpenOptions}; use std::io::{self, Read}; use std::os::windows::io::AsRawHandle; use std::path::{Path, PathBuf}; use std::time::Duration; pub type Writer = File; extern "system" { fn PeekNamedPipe( pipe: *mut c_void, buffer: *mut c_void, buffer_size: u32, bytes_read: *mut u32, total_bytes_avail: *mut u32, bytes_left_this_message: *mut u32, ) -> i32; } /// Reader half of the pipe. /// /// A blocking `ReadFile` on a synchronous pipe handle also blocks any /// `WriteFile` on a clone of that handle, which would deadlock the player. /// So we poll with `PeekNamedPipe` and only read when data is waiting. pub struct Reader(File); impl Read for Reader { fn read(&mut self, buf: &mut [u8]) -> io::Result { if buf.is_empty() { return Ok(0); } loop { let mut avail: u32 = 0; let ok = unsafe { PeekNamedPipe( self.0.as_raw_handle() as *mut c_void, std::ptr::null_mut(), 0, std::ptr::null_mut(), &mut avail, std::ptr::null_mut(), ) }; if ok == 0 { // mpv exited / pipe closed: report EOF. return Ok(0); } if avail > 0 { let n = (avail as usize).min(buf.len()); return self.0.read(&mut buf[..n]); } std::thread::sleep(Duration::from_millis(10)); } } } pub fn endpoint() -> PathBuf { PathBuf::from(format!(r"\\.\pipe\drift-{}", std::process::id())) } pub fn connect(path: &Path) -> io::Result<(Reader, Writer)> { let file = OpenOptions::new().read(true).write(true).open(path)?; let reader = file.try_clone()?; Ok((Reader(reader), file)) } pub fn cleanup(_path: &Path) { // Named pipes disappear when mpv exits. } } #[derive(Debug)] pub enum PlayerEvent { Position(Option), Duration(Option), Paused(bool), Volume(f64), FileLoaded, EndOfFile, LoadError(String), Died, } pub struct Player { child: Child, writer: ipc::Writer, sock_path: PathBuf, } impl Player { pub fn spawn(volume: u8, on_event: impl Fn(PlayerEvent) + Send + 'static) -> Result { let sock_path = ipc::endpoint(); let mut cmd = Command::new("mpv"); cmd.args([ "--no-config", "--idle=yes", "--no-video", "--no-terminal", "--really-quiet", "--audio-display=no", "--force-window=no", "--ytdl=no", "--cache=yes", "--demuxer-max-bytes=32MiB", ]) .arg(format!("--volume={volume}")) .arg(format!("--input-ipc-server={}", sock_path.display())); if let Ok(extra) = std::env::var("DRIFT_MPV_ARGS").or_else(|_| std::env::var("YTM_MPV_ARGS")) { cmd.args(extra.split_whitespace()); } cmd.stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()); crate::sys::no_window(&mut cmd); // Linux only: ask the kernel to SIGTERM mpv if drift dies unexpectedly. #[cfg(target_os = "linux")] unsafe { use std::os::unix::process::CommandExt; cmd.pre_exec(|| { libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM); Ok(()) }); } let mut child = cmd.spawn().map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { anyhow!("mpv not found — install it (see README)") } else { anyhow!("failed to start mpv: {e}") } })?; // Windows: tie mpv's lifetime to ours. crate::sys::kill_on_exit(&child); let start = Instant::now(); let (reader, writer) = loop { match ipc::connect(&sock_path) { Ok(s) => break s, Err(e) => { if let Ok(Some(status)) = child.try_wait() { bail!("mpv exited immediately ({status})"); } if start.elapsed() > Duration::from_secs(6) { let _ = child.kill(); let _ = child.wait(); bail!("timed out connecting to mpv IPC socket: {e}"); } std::thread::sleep(Duration::from_millis(40)); } } }; let mut player = Player { child, writer, sock_path, }; for (id, prop) in [ (1, "time-pos"), (2, "duration"), (3, "pause"), (4, "volume"), ] { player.send(json!(["observe_property", id, prop]))?; } player.send(json!(["request_log_messages", "error"]))?; std::thread::spawn(move || { let mut reason: Option = None; for line in BufReader::new(reader).lines() { let Ok(line) = line else { break }; let Ok(v) = serde_json::from_str::(&line) else { continue; }; match v.get("event").and_then(Value::as_str) { Some("log-message") => { let prefix = v.get("prefix").and_then(Value::as_str).unwrap_or(""); let text = v.get("text").and_then(Value::as_str).unwrap_or("").trim(); match text.strip_prefix("ERROR: ") { Some(t) if prefix == "ytdl_hook" => reason = Some(ytdl_reason(t)), _ if reason.is_none() && !text.is_empty() && !text.starts_with("Failed to open") => { reason = Some(text.lines().next().unwrap_or(text).to_string()); } _ => {} } continue; } Some("start-file") => reason = None, _ => {} } let Some(mut ev) = parse_event(&v) else { continue; }; match &mut ev { PlayerEvent::LoadError(msg) => { if let Some(r) = reason.take() { *msg = r; } } PlayerEvent::FileLoaded => reason = None, _ => {} } on_event(ev); } on_event(PlayerEvent::Died); }); Ok(player) } fn send(&mut self, command: Value) -> Result<()> { let mut line = serde_json::to_string(&json!({ "command": command }))?; line.push('\n'); self.writer .write_all(line.as_bytes()) .and_then(|_| self.writer.flush()) .map_err(|e| anyhow!("lost connection to mpv: {e}")) } pub fn is_alive(&mut self) -> bool { matches!(self.child.try_wait(), Ok(None)) } pub fn load(&mut self, path: &str) -> Result<()> { self.send(json!(["set_property", "pause", false]))?; self.send(json!(["loadfile", path, "replace"])) } pub fn set_pause(&mut self, pause: bool) -> Result<()> { self.send(json!(["set_property", "pause", pause])) } pub fn set_volume(&mut self, vol: u8) -> Result<()> { self.send(json!(["set_property", "volume", vol])) } pub fn seek_relative(&mut self, secs: f64) -> Result<()> { self.send(json!(["seek", secs, "relative"])) } pub fn seek_absolute(&mut self, secs: f64) -> Result<()> { self.send(json!(["seek", secs, "absolute"])) } pub fn stop(&mut self) -> Result<()> { self.send(json!(["stop"])) } } impl Drop for Player { fn drop(&mut self) { let _ = self.send(json!(["quit"])); for _ in 0..10 { if matches!(self.child.try_wait(), Ok(Some(_))) { break; } std::thread::sleep(Duration::from_millis(20)); } let _ = self.child.kill(); let _ = self.child.wait(); ipc::cleanup(&self.sock_path); } } const KEEP_FILES: usize = 40; fn audio_dir() -> PathBuf { dirs::cache_dir() .unwrap_or_else(std::env::temp_dir) .join("drift") .join("audio") } pub fn cached_audio(id: &str) -> Option { std::fs::read_dir(audio_dir()) .ok()? .flatten() .map(|e| e.path()) .find(|p| p.file_stem().and_then(|s| s.to_str()) == Some(id)) } pub fn fetch_audio(id: &str) -> Result { if let Some(path) = cached_audio(id) { return Ok(path); } let dir = audio_dir(); std::fs::create_dir_all(&dir).map_err(|e| format!("can't create {}: {e}", dir.display()))?; let mut cmd = Command::new("yt-dlp"); cmd.args(["-f", "bestaudio/best", "--no-playlist", "--no-warnings", "--no-progress", "--no-mtime"]) .arg("-P") .arg(&dir) .args(["-o", "%(id)s.%(ext)s"]) .arg(format!("https://music.youtube.com/watch?v={id}")) .stdin(Stdio::null()) .stdout(Stdio::null()); crate::sys::no_window(&mut cmd); let out = cmd .output() .map_err(|e| format!("could not run yt-dlp ({e}); is it installed and on PATH?"))?; if let Some(path) = cached_audio(id).filter(|_| out.status.success()) { prune(&dir); return Ok(path); } let stderr = String::from_utf8_lossy(&out.stderr); Err(stderr .lines() .rev() .find_map(|l| l.strip_prefix("ERROR: ")) .map(ytdl_reason) .unwrap_or_else(|| "yt-dlp downloaded nothing".into())) } fn prune(dir: &std::path::Path) { let Ok(entries) = std::fs::read_dir(dir) else { return; }; let mut files: Vec<_> = entries .flatten() .filter_map(|e| Some((e.metadata().ok()?.modified().ok()?, e.path()))) .collect(); if files.len() > KEEP_FILES { files.sort(); for (_, path) in &files[..files.len() - KEEP_FILES] { let _ = std::fs::remove_file(path); } } } pub fn ytdl_reason(text: &str) -> String { let text = text.split("; please report").next().unwrap_or(text).trim(); let text = match text.strip_prefix('[').and_then(|t| t.split_once("] ")) { Some((_, rest)) => match rest.split_once(": ") { Some((id, msg)) if id.len() == 11 && !id.contains(' ') => msg, _ => rest, }, None => text, }; text.trim().to_string() } fn parse_event(v: &Value) -> Option { match v.get("event")?.as_str()? { "property-change" => { let data = v.get("data"); match v.get("name")?.as_str()? { "time-pos" => Some(PlayerEvent::Position(data.and_then(Value::as_f64))), "duration" => Some(PlayerEvent::Duration(data.and_then(Value::as_f64))), "pause" => Some(PlayerEvent::Paused( data.and_then(Value::as_bool).unwrap_or(false), )), "volume" => data.and_then(Value::as_f64).map(PlayerEvent::Volume), _ => None, } } "file-loaded" => Some(PlayerEvent::FileLoaded), "end-file" => match v.get("reason").and_then(Value::as_str) { Some("eof") => Some(PlayerEvent::EndOfFile), Some("error") => Some(PlayerEvent::LoadError( v.get("file_error") .and_then(Value::as_str) .unwrap_or("playback failed") .to_string(), )), _ => None, }, _ => None, } } #[cfg(test)] mod tests { use super::*; #[test] #[ignore] fn mpv_roundtrip() { use std::sync::mpsc; let wav = std::env::temp_dir().join("drift-test.wav"); let samples: Vec = (0..16_000u32) .map(|i| if (i / 20) % 2 == 0 { 140 } else { 116 }) .collect(); let mut data = Vec::new(); data.extend_from_slice(b"RIFF"); data.extend_from_slice(&(36 + samples.len() as u32).to_le_bytes()); data.extend_from_slice(b"WAVEfmt "); data.extend_from_slice(&16u32.to_le_bytes()); data.extend_from_slice(&1u16.to_le_bytes()); data.extend_from_slice(&1u16.to_le_bytes()); data.extend_from_slice(&8000u32.to_le_bytes()); data.extend_from_slice(&8000u32.to_le_bytes()); data.extend_from_slice(&1u16.to_le_bytes()); data.extend_from_slice(&8u16.to_le_bytes()); data.extend_from_slice(b"data"); data.extend_from_slice(&(samples.len() as u32).to_le_bytes()); data.extend_from_slice(&samples); std::fs::write(&wav, data).unwrap(); std::env::set_var("DRIFT_MPV_ARGS", "--ao=null"); let (tx, rx) = mpsc::channel(); let mut p = Player::spawn(50, move |e| { let _ = tx.send(e); }) .expect("spawn mpv"); p.load(wav.to_str().unwrap()).unwrap(); let (mut loaded, mut eof) = (false, false); let deadline = Instant::now() + Duration::from_secs(10); while Instant::now() < deadline && !eof { match rx.recv_timeout(Duration::from_millis(200)) { Ok(PlayerEvent::FileLoaded) => loaded = true, Ok(PlayerEvent::EndOfFile) => eof = true, _ => {} } } assert!(loaded, "file-loaded not seen"); assert!(eof, "end-of-file not seen"); } #[test] fn ytdl_reason_is_trimmed() { assert_eq!( ytdl_reason("[youtube] dQw4w9WgXcQ: Sign in to confirm you're not a bot. Use --cookies. \n"), "Sign in to confirm you're not a bot. Use --cookies." ); assert_eq!( ytdl_reason("[youtube] dQw4w9WgXcQ: Failed to extract any player response; please report this issue on https://x"), "Failed to extract any player response" ); assert_eq!(ytdl_reason("Unsupported URL: foo"), "Unsupported URL: foo"); } #[test] fn event_parsing() { let v: Value = serde_json::from_str( r#"{"event":"property-change","id":1,"name":"time-pos","data":12.5}"#, ) .unwrap(); assert!(matches!(parse_event(&v), Some(PlayerEvent::Position(Some(p))) if p == 12.5)); let v: Value = serde_json::from_str(r#"{"event":"end-file","reason":"eof"}"#).unwrap(); assert!(matches!(parse_event(&v), Some(PlayerEvent::EndOfFile))); let v: Value = serde_json::from_str(r#"{"event":"end-file","reason":"stop"}"#).unwrap(); assert!(parse_event(&v).is_none()); let v: Value = serde_json::from_str(r#"{"error":"success","request_id":0}"#).unwrap(); assert!(parse_event(&v).is_none()); } }