use std::path::Path;
use std::process::Stdio;
use std::time::Duration;
use anyhow::{Context, Result};
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::oneshot;
use crate::chapters::Chapter;
use crate::tools::{self, Tools};
use crate::ytdlp::Streams;
const QUIT_GRACE: Duration = Duration::from_secs(3);
#[derive(Debug)]
pub struct PlayerEvent {
pub play: u64,
pub kind: PlayerEventKind,
}
#[derive(Debug, PartialEq)]
#[cfg_attr(not(unix), allow(dead_code))]
pub enum PlayerEventKind {
Position(f64),
Duration(f64),
Paused(bool),
Ended,
Exited,
}
pub struct Player {
pub play: u64,
controllable: bool,
#[cfg(unix)]
control: Option<tokio::sync::mpsc::UnboundedSender<String>>,
stop: Option<oneshot::Sender<()>>,
}
pub struct Start<'a> {
pub mpv: &'a Path,
pub tools: &'a Tools,
pub streams: &'a Streams,
pub title: &'a str,
pub start_at: Option<f64>,
pub audio_only: bool,
pub subtitles: Option<&'a str>,
pub subtitles_shown: bool,
}
impl Player {
pub fn start(
play: u64,
start: Start,
tx: UnboundedSender<crate::app::AppEvent>,
) -> Result<Self> {
let mut command = tokio::process::Command::new(start.mpv);
command
.args(args(&start))
.env_clear()
.envs(tools::child_env(start.tools))
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.kill_on_drop(true);
if let Some(home) = dirs::home_dir() {
command.current_dir(home);
}
#[cfg(unix)]
let (child, control) = {
use std::os::fd::AsRawFd;
let (ours, theirs) = std::os::unix::net::UnixStream::pair()?;
let fd = theirs.as_raw_fd();
unsafe {
command.pre_exec(move || {
let ok = if fd == 3 {
libc::fcntl(3, libc::F_SETFD, 0) != -1
} else {
libc::dup2(fd, 3) != -1
};
if !ok {
return Err(std::io::Error::last_os_error());
}
Ok(())
});
}
let child = command.spawn().context("cannot start mpv")?;
drop(theirs);
ours.set_nonblocking(true)?;
let stream = tokio::net::UnixStream::from_std(ours)?;
let (control_tx, control_rx) = tokio::sync::mpsc::unbounded_channel();
spawn_ipc(play, stream, control_rx, tx.clone());
for line in commands(&start) {
let _ = control_tx.send(line);
}
(child, control_tx)
};
#[cfg(not(unix))]
let child = command.spawn().context("cannot start mpv")?;
let (stop_tx, stop_rx) = oneshot::channel();
tokio::spawn(own(play, child, stop_rx, tx));
Ok(Self {
play,
controllable: cfg!(unix),
#[cfg(unix)]
control: Some(control),
stop: Some(stop_tx),
})
}
pub fn detached(play: u64) -> Self {
Self {
play,
controllable: cfg!(unix) || cfg!(test),
#[cfg(unix)]
control: None,
stop: None,
}
}
pub fn has_mpv(&self) -> bool {
self.stop.is_some()
}
pub fn controllable(&self) -> bool {
self.controllable
}
#[cfg(test)]
pub fn set_controllable(&mut self, controllable: bool) {
self.controllable = controllable;
}
fn send(&self, command: serde_json::Value) {
#[cfg(unix)]
if let Some(control) = &self.control {
let _ = control.send(line(command));
}
#[cfg(not(unix))]
let _ = command;
}
pub fn toggle_pause(&self) {
self.send(serde_json::json!(["cycle", "pause"]));
}
pub fn seek(&self, seconds: i32) {
self.send(serde_json::json!(["seek", seconds, "relative"]));
}
pub fn seek_to(&self, seconds: f64) {
if seconds.is_finite() && seconds >= 0.0 {
self.send(serde_json::json!(["seek", seconds, "absolute+exact"]));
}
}
pub fn show_subtitles(&self, shown: bool) {
self.send(serde_json::json!(["set_property", "sub-visibility", shown]));
}
pub fn set_chapters(&self, chapters: &[Chapter]) {
self.send(chapters_command(chapters));
}
pub fn stop(&mut self) {
self.send(serde_json::json!(["quit"]));
#[cfg(unix)]
{
self.control = None;
}
if let Some(stop) = self.stop.take() {
let _ = stop.send(());
}
}
}
impl Drop for Player {
fn drop(&mut self) {
self.stop();
}
}
async fn own(
play: u64,
mut child: tokio::process::Child,
stop: oneshot::Receiver<()>,
tx: UnboundedSender<crate::app::AppEvent>,
) {
tokio::select! {
_ = child.wait() => {}
_ = stop => {
let grace = if cfg!(unix) { QUIT_GRACE } else { Duration::ZERO };
if tokio::time::timeout(grace, child.wait()).await.is_err() {
let _ = child.start_kill();
let _ = child.wait().await;
}
}
}
let _ = tx.send(crate::app::AppEvent::Player(PlayerEvent {
play,
kind: PlayerEventKind::Exited,
}));
}
fn args(start: &Start) -> Vec<String> {
let mut args: Vec<String> = [
"--no-config",
"--load-scripts=no",
"--terminal=no",
"--ytdl=no",
"--tls-verify=yes",
"--keep-open=no",
"--save-position-on-quit=no",
"--resume-playback=no",
"--hwdec=auto-safe",
]
.map(String::from)
.into();
if start.audio_only {
args.push("--no-video".into());
args.push("--force-window=no".into());
} else {
args.push("--force-window=immediate".into());
}
if cfg!(unix) {
args.push("--idle=once".into());
args.push("--input-ipc-client=fd://3".into());
return args;
}
args.push("--idle=no".into());
for (name, value) in settings(start) {
args.push(format!("--{name}={value}"));
}
if let Some(audio) = &start.streams.audio {
args.push(format!("--audio-file={audio}"));
}
if let Some(subtitles) = start.subtitles {
args.push(format!("--sub-file={subtitles}"));
}
args.push("--".into());
args.push(start.streams.video.clone());
args
}
fn settings(start: &Start) -> Vec<(&'static str, String)> {
let title = crate::video::one_line(start.title, 150);
let mut settings = vec![
("force-media-title", title.clone()),
("title", format!("{} — tuitube", title.replace('$', "$$"))),
];
if let Some(ua) = &start.streams.user_agent {
settings.push(("user-agent", crate::video::one_line(ua, 300)));
}
if let Some(at) = start.start_at.filter(|s| s.is_finite() && *s > 0.0) {
settings.push(("start", format!("{at:.0}")));
}
if start.subtitles.is_some() {
let shown = if start.subtitles_shown { "yes" } else { "no" };
settings.push(("sub-visibility", shown.into()));
}
settings
}
fn chapters_command(chapters: &[Chapter]) -> serde_json::Value {
let list: Vec<serde_json::Value> = chapters
.iter()
.map(|c| serde_json::json!({ "title": c.title, "time": c.start }))
.collect();
serde_json::json!(["set_property", "chapter-list", list])
}
#[cfg_attr(not(unix), allow(dead_code))]
fn line(command: serde_json::Value) -> String {
format!("{}\n", serde_json::json!({ "command": command }))
}
#[cfg_attr(not(unix), allow(dead_code))]
fn commands(start: &Start) -> Vec<String> {
let mut commands: Vec<String> = [(1, "time-pos"), (2, "duration"), (3, "pause")]
.into_iter()
.map(|(id, property)| line(serde_json::json!(["observe_property", id, property])))
.collect();
for (name, value) in settings(start) {
commands.push(line(serde_json::json!(["set", name, value])));
}
if let Some(audio) = &start.streams.audio {
commands.push(line(serde_json::json!([
"change-list",
"audio-files",
"append",
audio
])));
}
if let Some(subtitles) = start.subtitles {
commands.push(line(serde_json::json!([
"change-list",
"sub-files",
"append",
subtitles
])));
}
commands.push(line(serde_json::json!([
"loadfile",
start.streams.video,
"replace"
])));
commands
}
#[cfg(unix)]
fn spawn_ipc(
play: u64,
stream: tokio::net::UnixStream,
mut control: tokio::sync::mpsc::UnboundedReceiver<String>,
tx: UnboundedSender<crate::app::AppEvent>,
) {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
let (read, mut write) = stream.into_split();
tokio::spawn(async move {
while let Some(line) = control.recv().await {
if write.write_all(line.as_bytes()).await.is_err() {
break;
}
}
});
tokio::spawn(async move {
let mut lines = BufReader::new(read).lines();
while let Ok(Some(line)) = lines.next_line().await {
if line.len() > 64 * 1024 {
continue;
}
if let Some(kind) = parse_event(&line) {
let _ = tx.send(crate::app::AppEvent::Player(PlayerEvent { play, kind }));
}
}
});
}
#[derive(serde::Deserialize)]
#[cfg(any(unix, test))]
struct IpcMessage {
event: Option<String>,
name: Option<String>,
data: Option<serde_json::Value>,
reason: Option<String>,
}
#[cfg(any(unix, test))]
fn parse_event(line: &str) -> Option<PlayerEventKind> {
let message: IpcMessage = serde_json::from_str(line).ok()?;
if message.event.as_deref() == Some("end-file") {
return (message.reason.as_deref() == Some("eof")).then_some(PlayerEventKind::Ended);
}
if message.event.as_deref() != Some("property-change") {
return None;
}
let data = message.data?;
match message.name.as_deref()? {
"time-pos" => data.as_f64().map(PlayerEventKind::Position),
"duration" => data.as_f64().map(PlayerEventKind::Duration),
"pause" => data.as_bool().map(PlayerEventKind::Paused),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn streams() -> Streams {
Streams {
video: "https://rr1.googlevideo.com/videoplayback?v=1".into(),
audio: Some("https://rr1.googlevideo.com/videoplayback?a=1,b=2".into()),
user_agent: Some("Mozilla/5.0\nX-Injected: yes".into()),
duration: Some(10.0),
chapters: Vec::new(),
quality: None,
info: None,
captions: Vec::new(),
}
}
fn start<'a>(tools: &'a Tools, streams: &'a Streams, title: &'a str) -> Start<'a> {
Start {
mpv: Path::new("/usr/bin/mpv"),
tools,
streams,
title,
start_at: Some(42.4),
audio_only: false,
subtitles: None,
subtitles_shown: false,
}
}
#[cfg(unix)]
#[test]
fn nothing_about_the_video_is_on_mpvs_command_line() {
let (tools, streams) = (Tools::default(), streams());
let args = args(&start(&tools, &streams, "My secret video"));
assert_eq!(args[0], "--no-config");
for flag in [
"--terminal=no",
"--ytdl=no",
"--tls-verify=yes",
"--idle=once",
] {
assert!(args.contains(&flag.to_string()), "{flag}");
}
let all = args.join(" ");
assert!(!all.contains("secret"), "{all}");
assert!(!all.contains("googlevideo"), "{all}");
assert!(!all.contains("Mozilla"), "{all}");
}
#[test]
fn the_socket_gets_the_video_as_json_strings() {
let (tools, streams) = (Tools::default(), streams());
let title = "\"]}{\"command\":[\"run\",\"sh\"]} ${path} --script=x";
let lines = commands(&start(&tools, &streams, title));
for line in &lines {
let parsed: serde_json::Value = serde_json::from_str(line.trim_end()).unwrap();
assert!(parsed["command"].is_array(), "{line}");
assert!(!line.trim_end().contains('\n'));
}
let set = |name: &str| {
lines
.iter()
.map(|l| serde_json::from_str::<serde_json::Value>(l).unwrap())
.find(|v| v["command"][0] == "set" && v["command"][1] == name)
.map(|v| v["command"][2].as_str().unwrap().to_string())
};
assert_eq!(
set("force-media-title").unwrap(),
title,
"kept whole, as one string"
);
assert!(
set("title").unwrap().contains("$${path}"),
"no property expansion"
);
assert_eq!(
set("user-agent").unwrap(),
"Mozilla/5.0 X-Injected: yes",
"one line"
);
assert_eq!(set("start").unwrap(), "42");
let last: serde_json::Value = serde_json::from_str(lines.last().unwrap()).unwrap();
assert_eq!(last["command"][0], "loadfile");
assert_eq!(last["command"][1], streams.video.as_str());
let audio: serde_json::Value = serde_json::from_str(&lines[lines.len() - 2]).unwrap();
assert_eq!(
audio["command"][2], "append",
"one item: commas don't split it"
);
}
#[test]
fn subtitles_go_in_with_the_video_shown_or_not() {
let (tools, streams) = (Tools::default(), streams());
let mut start = start(&tools, &streams, "A video");
let vtt = "https://www.youtube.com/api/timedtext?v=x&lang=en,fr&fmt=vtt";
start.subtitles = Some(vtt);
let lines: Vec<serde_json::Value> = commands(&start)
.iter()
.map(|l| serde_json::from_str(l).unwrap())
.collect();
let sub = lines
.iter()
.find(|v| v["command"][1] == "sub-files")
.expect("the subtitles");
assert_eq!(
sub["command"],
serde_json::json!(["change-list", "sub-files", "append", vtt]),
"one item: the comma doesn't split it"
);
let shown = lines
.iter()
.find(|v| v["command"][0] == "set" && v["command"][1] == "sub-visibility")
.unwrap();
assert_eq!(shown["command"][2], "no");
assert_eq!(lines.last().unwrap()["command"][0], "loadfile");
start.subtitles = None;
let none = commands(&start).join("");
assert!(!none.contains("sub-"), "{none}");
}
#[test]
fn chapters_go_to_mpv_as_one_command_of_json_strings() {
let chapters = [
Chapter {
start: 0.0,
title: "Intro".into(),
},
Chapter {
start: 83.5,
title: "\"]}\n{\"command\":[\"run\",\"sh\"]} ${path}".into(),
},
];
let sent = line(chapters_command(&chapters));
assert_eq!(sent.matches('\n').count(), 1, "one line: {sent}");
let parsed: serde_json::Value = serde_json::from_str(sent.trim_end()).unwrap();
let command = &parsed["command"];
assert_eq!(command[0], "set_property");
assert_eq!(command[1], "chapter-list");
assert_eq!(command[2][1]["time"], 83.5);
assert_eq!(
command[2][1]["title"],
chapters[1].title.as_str(),
"kept whole"
);
assert_eq!(command.as_array().unwrap().len(), 3);
}
#[test]
fn only_followed_properties_become_events() {
let pos = r#"{"event":"property-change","id":1,"name":"time-pos","data":12.5}"#;
assert_eq!(parse_event(pos), Some(PlayerEventKind::Position(12.5)));
let pause = r#"{"event":"property-change","id":3,"name":"pause","data":true}"#;
assert_eq!(parse_event(pause), Some(PlayerEventKind::Paused(true)));
assert_eq!(
parse_event(r#"{"event":"property-change","name":"time-pos","data":null}"#),
None
);
assert_eq!(parse_event(r#"{"request_id":0,"error":"success"}"#), None);
assert_eq!(parse_event("not json"), None);
}
#[test]
fn only_playing_to_the_end_counts_as_ended() {
let end = |reason: &str| {
parse_event(&format!(
r#"{{"event":"end-file","reason":"{reason}","playlist_entry_id":1}}"#
))
};
assert_eq!(end("eof"), Some(PlayerEventKind::Ended));
for reason in ["quit", "stop", "error", "redirect", "unknown"] {
assert_eq!(end(reason), None, "{reason}");
}
assert_eq!(parse_event(r#"{"event":"end-file"}"#), None);
}
}
#[cfg(all(test, unix))]
mod live {
use super::*;
use crate::app::AppEvent;
#[tokio::test]
#[ignore = "runs mpv"]
async fn live_mpv_plays_what_the_socket_sends_and_stops() {
let tools = Tools::find(None, None, None);
let mpv = tools.mpv.clone().expect("mpv is installed");
let streams = Streams {
video: "av://lavfi:anullsrc=d=5".into(),
audio: Some("av://lavfi:anullsrc=d=5".into()),
user_agent: Some("tuitube-test".into()),
duration: None,
chapters: Vec::new(),
quality: None,
info: None,
captions: Vec::new(),
};
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let start = Start {
mpv: &mpv,
tools: &tools,
streams: &streams,
title: "silence, with \"quotes\" and ${path}",
start_at: Some(1.0),
audio_only: true,
subtitles: None,
subtitles_shown: false,
};
let mut player = Player::start(7, start, tx).unwrap();
let mut positions = Vec::new();
let mut exited = false;
let mut ended = false;
let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
while let Ok(Some(event)) = tokio::time::timeout_at(deadline, rx.recv()).await {
let AppEvent::Player(event) = event else {
continue;
};
assert_eq!(event.play, 7);
match event.kind {
PlayerEventKind::Position(at) => {
positions.push(at);
if positions.len() == 3 {
player.stop();
}
}
PlayerEventKind::Ended => ended = true,
PlayerEventKind::Exited => {
exited = true;
break;
}
_ => {}
}
}
assert!(
!positions.is_empty(),
"nothing played: the socket's file didn't load"
);
assert!(positions[0] >= 0.9, "started at --start: {positions:?}");
assert!(exited, "mpv didn't quit");
assert!(!ended, "stopped isn't played to the end");
}
#[tokio::test]
#[ignore = "runs mpv"]
async fn live_mpv_seeks_to_where_a_skipped_segment_ends() {
let tools = Tools::find(None, None, None);
let mpv = tools.mpv.clone().expect("mpv is installed");
let streams = Streams {
video: "av://lavfi:anullsrc=d=10".into(),
audio: None,
user_agent: None,
duration: None,
chapters: Vec::new(),
quality: None,
info: None,
captions: Vec::new(),
};
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let start = Start {
mpv: &mpv,
tools: &tools,
streams: &streams,
title: "ten seconds of silence",
start_at: None,
audio_only: true,
subtitles: None,
subtitles_shown: false,
};
let mut player = Player::start(9, start, tx).unwrap();
let mut sought = None;
let mut landed = None;
let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
while let Ok(Some(event)) = tokio::time::timeout_at(deadline, rx.recv()).await {
let AppEvent::Player(event) = event else {
continue;
};
match event.kind {
PlayerEventKind::Position(at) if sought.is_none() && at < 1.0 => {
player.seek_to(6.5);
sought = Some(std::time::Instant::now());
}
PlayerEventKind::Position(at) if at >= 6.5 && landed.is_none() => {
landed = sought.map(|t| (at, t.elapsed()));
player.stop();
}
PlayerEventKind::Exited => break,
_ => {}
}
}
let (at, after) = landed.expect("the seek didn't happen");
assert!(after < Duration::from_secs(3), "{after:?}");
assert!(at < 8.0, "landed at {at}");
}
#[tokio::test]
#[ignore = "runs mpv"]
async fn live_mpv_says_when_a_video_plays_to_its_end() {
let tools = Tools::find(None, None, None);
let mpv = tools.mpv.clone().expect("mpv is installed");
let streams = Streams {
video: "av://lavfi:anullsrc=d=2".into(),
audio: None,
user_agent: None,
duration: None,
chapters: Vec::new(),
quality: None,
info: None,
captions: Vec::new(),
};
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let start = Start {
mpv: &mpv,
tools: &tools,
streams: &streams,
title: "two seconds of silence",
start_at: None,
audio_only: true,
subtitles: None,
subtitles_shown: false,
};
let _player = Player::start(8, start, tx).unwrap();
let mut kinds = Vec::new();
let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
while let Ok(Some(event)) = tokio::time::timeout_at(deadline, rx.recv()).await {
if let AppEvent::Player(event) = event {
match event.kind {
PlayerEventKind::Ended | PlayerEventKind::Exited => kinds.push(event.kind),
_ => {}
}
if kinds.last() == Some(&PlayerEventKind::Exited) {
break;
}
}
}
assert_eq!(
kinds,
[PlayerEventKind::Ended, PlayerEventKind::Exited],
"ended, then quit by itself"
);
}
}