WIP: centralize mpv broker

.envrc

@@ -0,0 +1 @@
+use flake

.gitignore

@@ -1 +1,2 @@
 /target
+result

Cargo.toml

@@ -2,6 +2,9 @@
 name = "greg-ng"
 version = "0.1.0"
 edition = "2021"
+license = "MIT"
+authors = ["oysteikt@pvv.ntnu.no"]
+readme = "README.md"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -11,9 +14,14 @@ axum = { version = "0.6.20", features = ["macros"] }
 clap = { version = "4.4.1", features = ["derive"] }
 env_logger = "0.10.0"
 log = "0.4.20"
-mpvipc = "1.3.0"
+mpvipc-async = { git = "https://git.pvv.ntnu.no/oysteikt/mpvipc-async.git", rev = "v0.1.0" }
 serde = { version = "1.0.188", features = ["derive"] }
 serde_json = "1.0.105"
 tokio = { version = "1.32.0", features = ["full"] }
 tower = { version = "0.4.13", features = ["full"] }
 tower-http = { version = "0.4.3", features = ["full"] }
+
+[profile.release]
+strip = true
+lto = true
+codegen-units = 1

README.md

@@ -2,23 +2,20 @@
 
 New implementation of https://github.com/Programvareverkstedet/grzegorz
 
-## Feature wishlist
+## Test it out
 
-- [ ] Feature parity with old grzegorz
-  - [X] Rest API
-  - [ ] Rest API docs
-  - [ ] Metadata fetcher
-  - [ ] Init mpv with image of grzegorz
-- [ ] Save playlists to machine
-- [ ] Cache playlist contents to disk
-- [ ] Expose service through mpd protocol
-- [ ] Users with playlists and songs (and auth?)
-- [ ] Some kind of fair scheduling for each user
-- [ ] Max time to avoid playlist songs
-- [ ] Expose video/media stream so others can listen at home
-- [ ] Syncope support >:)
-- [ ] Jitsi support >:)))
-- [ ] Show other media while playing music, like grafana or bustimes
-- [ ] Soft shuffle
-- [ ] Libre.fm integration
-- [ ] Karaoke mode lmao
+```sh
+# NixOS
+nix run "git+https://git.pvv.ntnu.no/Projects/greg-ng#" -- --mpv-socket-path /tmp/mpv.sock
+
+# Other (after git clone and rust toolchain has been set up)
+cargo run -- --mpv-socket-path /tmp/mpv.sock
+```
+
+See also https://git.pvv.ntnu.no/Projects/grzegorz-clients for frontend alternatives
+
+## Debugging
+
+```sh
+RUST_LOG=greg_ng=trace,mpvipc=trace cargo run -- --mpv-socket-path /tmp/mpv.sock
+```

default.nix

@@ -0,0 +1,34 @@
+{
+  lib
+, fetchFromGitHub
+, rustPlatform
+, makeWrapper
+, mpv
+}:
+
+rustPlatform.buildRustPackage rec {
+  pname = "greg-ng";
+  version = "0.1.0";
+  src = lib.cleanSource ./.;
+
+  nativeBuildInputs = [ makeWrapper ];
+
+  cargoLock = {
+    lockFile = ./Cargo.lock;
+    outputHashes = {
+      "mpvipc-async-0.1.0" = "sha256-2TQ2d4q9/DTxTZe9kOAoKBhsmegRZw32x3G2hbluS6U=";
+    };
+  };
+
+  postInstall = ''
+    wrapProgram $out/bin/greg-ng \
+      --prefix PATH : '${lib.makeBinPath [ mpv ]}'
+  '';
+
+  meta = with lib; {
+    license = licenses.mit;
+    maintainers = with maintainers; [ h7x4 ];
+    platforms = platforms.linux ++ platforms.darwin;
+    mainProgram = "greg-ng";
+  };
+}

flake.lock

@@ -0,0 +1,48 @@
+{
+  "nodes": {
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1722421184,
+        "narHash": "sha256-/DJBI6trCeVnasdjUo9pbnodCLZcFqnVZiLUfqLH4jA=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "9f918d616c5321ad374ae6cb5ea89c9e04bf3e58",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixos-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "nixpkgs": "nixpkgs",
+        "rust-overlay": "rust-overlay"
+      }
+    },
+    "rust-overlay": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1722651535,
+        "narHash": "sha256-2uRmNwxe3CO5h7PfvqXrRe8OplXaEdwhqOUtaF13rpU=",
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "rev": "56d83ca6f3c557647476f3720426a7615c22b860",
+        "type": "github"
+      },
+      "original": {
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "type": "github"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,59 @@
+{
+  inputs = {
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+
+    rust-overlay.url = "github:oxalica/rust-overlay";
+    rust-overlay.inputs.nixpkgs.follows = "nixpkgs";
+  };
+
+  outputs = { self, nixpkgs, rust-overlay }:
+  let
+    inherit (nixpkgs) lib;
+
+    systems = [
+      "x86_64-linux"
+      "aarch64-linux"
+      "x86_64-darwin"
+      "aarch64-darwin"
+      "armv7l-linux"
+    ];
+
+    forAllSystems = f: lib.genAttrs systems (system: let
+      pkgs = import nixpkgs {
+        inherit system;
+        overlays = [
+          (import rust-overlay)
+        ];
+      };
+
+      rust-bin = rust-overlay.lib.mkRustBin { } pkgs.buildPackages;
+      toolchain = rust-bin.stable.latest.default;
+    in f system pkgs toolchain);
+  in {
+    apps = forAllSystems (system: pkgs: _: {
+      default = self.apps.${system}.greg-ng;
+      greg-ng = let
+        package = self.packages.${system}.greg-ng;
+      in {
+        type = "app";
+        program = lib.getExe package;
+      };
+    });
+
+    devShells = forAllSystems (system: pkgs: toolchain: {
+      default = pkgs.mkShell {
+        nativeBuildInputs = [
+          toolchain
+          pkgs.mpv
+        ];
+
+        RUST_SRC_PATH = "${toolchain}/lib/rustlib/src/rust/library";
+      };
+    });
+
+    packages = forAllSystems (system: pkgs: _: {
+      default = self.packages.${system}.greg-ng;
+      greg-ng = pkgs.callPackage ./default.nix { };
+    });
+  };
+}

src/api/base.rs

@@ -1,62 +1,61 @@
-use std::sync::Arc;
-
-use log::trace;
-use mpvipc::{
-    Mpv, NumberChangeOptions, PlaylistAddOptions, PlaylistAddTypeOptions, SeekOptions, Switch,
+use mpvipc_async::{
+    LoopProperty, Mpv, MpvExt, NumberChangeOptions, PlaylistAddOptions, PlaylistAddTypeOptions,
+    SeekOptions, Switch,
 };
 use serde_json::{json, Value};
-use tokio::sync::Mutex;
 
 /// Add item to playlist
-pub async fn loadfile(mpv: Arc<Mutex<Mpv>>, path: &str) -> anyhow::Result<()> {
-    trace!("api::loadfile({:?})", path);
-    mpv.lock().await.playlist_add(
+pub async fn loadfile(mpv: Mpv, path: &str) -> anyhow::Result<()> {
+    log::trace!("api::loadfile({:?})", path);
+    mpv.playlist_add(
         path,
         PlaylistAddTypeOptions::File,
         PlaylistAddOptions::Append,
-    )?;
+    )
+    .await?;
 
     Ok(())
 }
 
 /// Check whether the player is paused or playing
-pub async fn play_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
-    trace!("api::play_get()");
-    let paused: bool = mpv.lock().await.get_property("pause")?;
+pub async fn play_get(mpv: Mpv) -> anyhow::Result<Value> {
+    log::trace!("api::play_get()");
+    let paused: bool = !mpv.is_playing().await?;
     Ok(json!(!paused))
 }
 
 /// Set whether the player is paused or playing
-pub async fn play_set(mpv: Arc<Mutex<Mpv>>, should_play: bool) -> anyhow::Result<()> {
-    trace!("api::play_set({:?})", should_play);
-    mpv.lock()
+pub async fn play_set(mpv: Mpv, should_play: bool) -> anyhow::Result<()> {
+    log::trace!("api::play_set({:?})", should_play);
+    mpv.set_playback(if should_play { Switch::On } else { Switch::Off })
         .await
-        .set_property("pause", !should_play)
         .map_err(|e| e.into())
 }
 
 /// Get the current player volume
-pub async fn volume_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
-    trace!("api::volume_get()");
-    let volume: f64 = mpv.lock().await.get_property("volume")?;
+pub async fn volume_get(mpv: Mpv) -> anyhow::Result<Value> {
+    log::trace!("api::volume_get()");
+    let volume: f64 = mpv.get_volume().await?;
     Ok(json!(volume))
 }
 
 /// Set the player volume
-pub async fn volume_set(mpv: Arc<Mutex<Mpv>>, value: f64) -> anyhow::Result<()> {
-    trace!("api::volume_set({:?})", value);
-    mpv.lock()
+pub async fn volume_set(mpv: Mpv, value: f64) -> anyhow::Result<()> {
+    log::trace!("api::volume_set({:?})", value);
+    mpv.set_volume(value, NumberChangeOptions::Absolute)
         .await
-        .set_volume(value, NumberChangeOptions::Absolute)
         .map_err(|e| e.into())
 }
 
 /// Get current playback position
-pub async fn time_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
-    trace!("api::time_get()");
-    let current: f64 = mpv.lock().await.get_property("time-pos")?;
-    let remaining: f64 = mpv.lock().await.get_property("time-remaining")?;
-    let total = current + remaining;
+pub async fn time_get(mpv: Mpv) -> anyhow::Result<Value> {
+    log::trace!("api::time_get()");
+    let current: Option<f64> = mpv.get_time_pos().await?;
+    let remaining: Option<f64> = mpv.get_time_remaining().await?;
+    let total = match (current, remaining) {
+        (Some(c), Some(r)) => Some(c + r),
+        (_, _) => None,
+    };
 
     Ok(json!({
         "current": current,
@@ -66,22 +65,16 @@ pub async fn time_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
 }
 
 /// Set playback position
-pub async fn time_set(
-    mpv: Arc<Mutex<Mpv>>,
-    pos: Option<f64>,
-    percent: Option<f64>,
-) -> anyhow::Result<()> {
-    trace!("api::time_set({:?}, {:?})", pos, percent);
+pub async fn time_set(mpv: Mpv, pos: Option<f64>, percent: Option<f64>) -> anyhow::Result<()> {
+    log::trace!("api::time_set({:?}, {:?})", pos, percent);
     if pos.is_some() && percent.is_some() {
         anyhow::bail!("pos and percent cannot be provided at the same time");
     }
 
     if let Some(pos) = pos {
-        mpv.lock().await.seek(pos, SeekOptions::Absolute)?;
+        mpv.seek(pos, SeekOptions::Absolute).await?;
     } else if let Some(percent) = percent {
-        mpv.lock()
-            .await
-            .seek(percent, SeekOptions::AbsolutePercent)?;
+        mpv.seek(percent, SeekOptions::AbsolutePercent).await?;
     } else {
         anyhow::bail!("Either pos or percent must be provided");
     };
@@ -90,10 +83,10 @@ pub async fn time_set(
 }
 
 /// Get the current playlist
-pub async fn playlist_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
-    trace!("api::playlist_get()");
-    let playlist: mpvipc::Playlist = mpv.lock().await.get_playlist()?;
-    let is_playing: bool = mpv.lock().await.get_property("pause")?;
+pub async fn playlist_get(mpv: Mpv) -> anyhow::Result<Value> {
+    log::trace!("api::playlist_get()");
+    let playlist: mpvipc_async::Playlist = mpv.get_playlist().await?;
+    let is_playing: bool = mpv.is_playing().await?;
 
     let items: Vec<Value> = playlist
         .0
@@ -104,7 +97,7 @@ pub async fn playlist_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
               "index": i,
               "current": item.current,
               "playing": is_playing,
-              "filename": item.filename,
+              "filename": item.title.as_ref().unwrap_or(&item.filename),
               "data": {
                 "fetching": true,
               }
@@ -116,74 +109,64 @@ pub async fn playlist_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
 }
 
 /// Skip to the next item in the playlist
-pub async fn playlist_next(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<()> {
-    trace!("api::playlist_next()");
-    mpv.lock().await.next().map_err(|e| e.into())
+pub async fn playlist_next(mpv: Mpv) -> anyhow::Result<()> {
+    log::trace!("api::playlist_next()");
+    mpv.next().await.map_err(|e| e.into())
 }
 
 /// Go back to the previous item in the playlist
-pub async fn playlist_previous(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<()> {
-    trace!("api::playlist_previous()");
-    mpv.lock().await.prev().map_err(|e| e.into())
+pub async fn playlist_previous(mpv: Mpv) -> anyhow::Result<()> {
+    log::trace!("api::playlist_previous()");
+    mpv.prev().await.map_err(|e| e.into())
 }
 
 /// Go chosen item in the playlist
-pub async fn playlist_goto(mpv: Arc<Mutex<Mpv>>, index: usize) -> anyhow::Result<()> {
-    trace!("api::playlist_goto({:?})", index);
-    mpv.lock()
-        .await
-        .playlist_play_id(index)
-        .map_err(|e| e.into())
+pub async fn playlist_goto(mpv: Mpv, index: usize) -> anyhow::Result<()> {
+    log::trace!("api::playlist_goto({:?})", index);
+    mpv.playlist_play_id(index).await.map_err(|e| e.into())
 }
 
 /// Clears the playlist
-pub async fn playlist_clear(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<()> {
-    trace!("api::playlist_clear()");
-    mpv.lock().await.playlist_clear().map_err(|e| e.into())
+pub async fn playlist_clear(mpv: Mpv) -> anyhow::Result<()> {
+    log::trace!("api::playlist_clear()");
+    mpv.playlist_clear().await.map_err(|e| e.into())
 }
 
 /// Remove an item from the playlist by index
-pub async fn playlist_remove(mpv: Arc<Mutex<Mpv>>, index: usize) -> anyhow::Result<()> {
-    trace!("api::playlist_remove({:?})", index);
-    mpv.lock()
-        .await
-        .playlist_remove_id(index)
-        .map_err(|e| e.into())
+pub async fn playlist_remove(mpv: Mpv, index: usize) -> anyhow::Result<()> {
+    log::trace!("api::playlist_remove({:?})", index);
+    mpv.playlist_remove_id(index).await.map_err(|e| e.into())
 }
 
 /// Move an item in the playlist from one index to another
-pub async fn playlist_move(mpv: Arc<Mutex<Mpv>>, from: usize, to: usize) -> anyhow::Result<()> {
-    trace!("api::playlist_move({:?}, {:?})", from, to);
-    mpv.lock()
-        .await
-        .playlist_move_id(from, to)
-        .map_err(|e| e.into())
+pub async fn playlist_move(mpv: Mpv, from: usize, to: usize) -> anyhow::Result<()> {
+    log::trace!("api::playlist_move({:?}, {:?})", from, to);
+    mpv.playlist_move_id(from, to).await.map_err(|e| e.into())
 }
 
 /// Shuffle the playlist
-pub async fn shuffle(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<()> {
-    trace!("api::shuffle()");
-    mpv.lock().await.playlist_shuffle().map_err(|e| e.into())
+pub async fn shuffle(mpv: Mpv) -> anyhow::Result<()> {
+    log::trace!("api::shuffle()");
+    mpv.playlist_shuffle().await.map_err(|e| e.into())
 }
 
 /// See whether it loops the playlist or not
-pub async fn playlist_get_looping(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
-    trace!("api::playlist_get_looping()");
-    let loop_playlist = mpv.lock().await.get_property_string("loop-playlist")? == "inf";
-    Ok(json!(loop_playlist))
+pub async fn playlist_get_looping(mpv: Mpv) -> anyhow::Result<Value> {
+    log::trace!("api::playlist_get_looping()");
+
+    let loop_status = match mpv.playlist_is_looping().await? {
+        LoopProperty::No => false,
+        LoopProperty::Inf => true,
+        LoopProperty::N(_) => true,
+    };
+
+    Ok(json!(loop_status))
 }
 
-pub async fn playlist_set_looping(mpv: Arc<Mutex<Mpv>>, r#loop: bool) -> anyhow::Result<()> {
-    trace!("api::playlist_set_looping({:?})", r#loop);
-    if r#loop {
-        mpv.lock()
-            .await
-            .set_loop_playlist(Switch::On)
-            .map_err(|e| e.into())
-    } else {
-        mpv.lock()
-            .await
-            .set_loop_playlist(Switch::Off)
-            .map_err(|e| e.into())
-    }
+pub async fn playlist_set_looping(mpv: Mpv, r#loop: bool) -> anyhow::Result<()> {
+    log::trace!("api::playlist_set_looping({:?})", r#loop);
+
+    mpv.set_loop_playlist(if r#loop { Switch::On } else { Switch::Off })
+        .await
+        .map_err(|e| e.into())
 }

src/api/rest_wrapper_v1.rs

@@ -1,5 +1,3 @@
-use std::sync::Arc;
-
 use axum::{
     extract::{Query, State},
     http::StatusCode,
@@ -7,13 +5,12 @@ use axum::{
     routing::{delete, get, post},
     Json, Router,
 };
-use mpvipc::Mpv;
+use mpvipc_async::Mpv;
 use serde_json::{json, Value};
-use tokio::sync::Mutex;
 
 use super::base;
 
-pub fn rest_api_routes(mpv: Arc<Mutex<Mpv>>) -> Router {
+pub fn rest_api_routes(mpv: Mpv) -> Router {
     Router::new()
         .route("/load", post(loadfile))
         .route("/play", get(play_get))
@@ -54,7 +51,7 @@ impl IntoResponse for RestResponse {
             Ok(value) => (StatusCode::OK, Json(value)).into_response(),
             Err(err) => (
                 StatusCode::INTERNAL_SERVER_ERROR,
-                Json(json!({ "error": err.to_string(), "success": false })),
+                Json(json!({ "error": err.to_string(), "errortext": err.to_string(), "success": false })),
             )
                 .into_response(),
         }
@@ -72,14 +69,11 @@ struct LoadFileArgs {
     path: String,
 }
 
-async fn loadfile(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
-    Query(query): Query<LoadFileArgs>,
-) -> RestResponse {
+async fn loadfile(State(mpv): State<Mpv>, Query(query): Query<LoadFileArgs>) -> RestResponse {
     base::loadfile(mpv, &query.path).await.into()
 }
 
-async fn play_get(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn play_get(State(mpv): State<Mpv>) -> RestResponse {
     base::play_get(mpv).await.into()
 }
 
@@ -88,15 +82,12 @@ struct PlaySetArgs {
     play: String,
 }
 
-async fn play_set(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
-    Query(query): Query<PlaySetArgs>,
-) -> RestResponse {
+async fn play_set(State(mpv): State<Mpv>, Query(query): Query<PlaySetArgs>) -> RestResponse {
     let play = query.play.to_lowercase() == "true";
     base::play_set(mpv, play).await.into()
 }
 
-async fn volume_get(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn volume_get(State(mpv): State<Mpv>) -> RestResponse {
     base::volume_get(mpv).await.into()
 }
 
@@ -105,14 +96,11 @@ struct VolumeSetArgs {
     volume: f64,
 }
 
-async fn volume_set(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
-    Query(query): Query<VolumeSetArgs>,
-) -> RestResponse {
+async fn volume_set(State(mpv): State<Mpv>, Query(query): Query<VolumeSetArgs>) -> RestResponse {
     base::volume_set(mpv, query.volume).await.into()
 }
 
-async fn time_get(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn time_get(State(mpv): State<Mpv>) -> RestResponse {
     base::time_get(mpv).await.into()
 }
 
@@ -122,22 +110,19 @@ struct TimeSetArgs {
     percent: Option<f64>,
 }
 
-async fn time_set(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
-    Query(query): Query<TimeSetArgs>,
-) -> RestResponse {
+async fn time_set(State(mpv): State<Mpv>, Query(query): Query<TimeSetArgs>) -> RestResponse {
     base::time_set(mpv, query.pos, query.percent).await.into()
 }
 
-async fn playlist_get(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn playlist_get(State(mpv): State<Mpv>) -> RestResponse {
     base::playlist_get(mpv).await.into()
 }
 
-async fn playlist_next(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn playlist_next(State(mpv): State<Mpv>) -> RestResponse {
     base::playlist_next(mpv).await.into()
 }
 
-async fn playlist_previous(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn playlist_previous(State(mpv): State<Mpv>) -> RestResponse {
     base::playlist_previous(mpv).await.into()
 }
 
@@ -147,7 +132,7 @@ struct PlaylistGotoArgs {
 }
 
 async fn playlist_goto(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
+    State(mpv): State<Mpv>,
     Query(query): Query<PlaylistGotoArgs>,
 ) -> RestResponse {
     base::playlist_goto(mpv, query.index).await.into()
@@ -159,7 +144,7 @@ struct PlaylistRemoveOrClearArgs {
 }
 
 async fn playlist_remove_or_clear(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
+    State(mpv): State<Mpv>,
     Query(query): Query<PlaylistRemoveOrClearArgs>,
 ) -> RestResponse {
     match query.index {
@@ -175,7 +160,7 @@ struct PlaylistMoveArgs {
 }
 
 async fn playlist_move(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
+    State(mpv): State<Mpv>,
     Query(query): Query<PlaylistMoveArgs>,
 ) -> RestResponse {
     base::playlist_move(mpv, query.index1, query.index2)
@@ -183,11 +168,11 @@ async fn playlist_move(
         .into()
 }
 
-async fn shuffle(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn shuffle(State(mpv): State<Mpv>) -> RestResponse {
     base::shuffle(mpv).await.into()
 }
 
-async fn playlist_get_looping(State(mpv): State<Arc<Mutex<Mpv>>>) -> RestResponse {
+async fn playlist_get_looping(State(mpv): State<Mpv>) -> RestResponse {
     base::playlist_get_looping(mpv).await.into()
 }
 
@@ -197,7 +182,7 @@ struct PlaylistSetLoopingArgs {
 }
 
 async fn playlist_set_looping(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
+    State(mpv): State<Mpv>,
     Query(query): Query<PlaylistSetLoopingArgs>,
 ) -> RestResponse {
     base::playlist_set_looping(mpv, query.r#loop).await.into()

src/main.rs

@@ -1,17 +1,13 @@
 use anyhow::Context;
 use axum::{Router, Server};
 use clap::Parser;
-use mpvipc::Mpv;
+use mpvipc_async::Mpv;
 use std::{
     fs::create_dir_all,
     net::{IpAddr, SocketAddr},
     path::Path,
-    sync::Arc,
-};
-use tokio::{
-    process::{Child, Command},
-    sync::Mutex,
 };
+use tokio::process::{Child, Command};
 
 mod api;
 
@@ -84,9 +80,9 @@ async fn connect_to_mpv(args: &MpvConnectionArgs) -> anyhow::Result<(Mpv, Option
                 .arg(format!("--input-ipc-server={}", &args.socket_path))
                 .arg("--idle")
                 .arg("--force-window")
-                // .arg("--fullscreen")
+                .arg("--fullscreen")
                 // .arg("--no-terminal")
-                // .arg("--load-unsafe-playlists")
+                .arg("--load-unsafe-playlists")
                 .arg("--keep-open") // Keep last frame of video on end of video
                 .spawn()
                 .context("Failed to start mpv")?,
@@ -112,7 +108,7 @@ async fn connect_to_mpv(args: &MpvConnectionArgs) -> anyhow::Result<(Mpv, Option
     }
 
     Ok((
-        Mpv::connect(&args.socket_path).context(format!(
+        Mpv::connect(&args.socket_path).await.context(format!(
             "Failed to connect to mpv socket: {}",
             &args.socket_path
         ))?,
@@ -146,25 +142,27 @@ async fn main() -> anyhow::Result<()> {
     let addr = SocketAddr::new(resolve(&args.host).await?, args.port);
     log::info!("Starting API on {}", addr);
 
-    let mpv = Arc::new(Mutex::new(mpv));
-    let app = Router::new().nest("/api", api::rest_api_routes(mpv));
+    let app = Router::new().nest("/api", api::rest_api_routes(mpv.clone()));
 
     if let Some(mut proc) = proc {
         tokio::select! {
-          exit_status = proc.wait() => {
-              log::warn!("mpv process exited with status: {}", exit_status?);
-          }
-          _ = tokio::signal::ctrl_c() => {
-              log::info!("Received Ctrl-C, exiting");
-              proc.kill().await?;
-          }
-          result = async {
+            exit_status = proc.wait() => {
+                log::warn!("mpv process exited with status: {}", exit_status?);
+                mpv.disconnect().await?;
+            }
+            _ = tokio::signal::ctrl_c() => {
+                log::info!("Received Ctrl-C, exiting");
+                mpv.disconnect().await?;
+                proc.kill().await?;
+            }
+            result = async {
               match Server::try_bind(&addr.clone()).context("Failed to bind server") {
                 Ok(server) => server.serve(app.into_make_service()).await.context("Failed to serve app"),
                 Err(err) => Err(err),
               }
             } => {
               log::info!("API server exited");
+              mpv.disconnect().await?;
               proc.kill().await?;
               result?;
             }
@@ -173,9 +171,11 @@ async fn main() -> anyhow::Result<()> {
         tokio::select! {
             _ = tokio::signal::ctrl_c() => {
                 log::info!("Received Ctrl-C, exiting");
+                mpv.disconnect().await?;
             }
             _ =  Server::bind(&addr.clone()).serve(app.into_make_service()) => {
                 log::info!("API server exited");
+                mpv.disconnect().await?;
             }
         }
     }

src/mpv_broker.rs

@@ -0,0 +1,154 @@
+use std::{fs::create_dir_all, path::Path};
+
+use anyhow::Context;
+use mpvipc_async::{Mpv, MpvCommand, Event as MpvEvent};
+use tokio::{process::{Child, Command}, sync::{broadcast::{Receiver as BroadcastReceiver, Sender as BroadcastSender}, mpsc::{Receiver as MpscReceiver, Sender as MpscSender}}};
+
+#[derive(Debug)]
+pub struct MpvBroker {
+    mpv: Mpv,
+    command_channel: MpscReceiver<MpvCommand>,
+    event_listeners: BroadcastSender<MpvEvent>,
+}
+
+impl MpvBroker {
+    pub fn new(
+        mpv: Mpv,
+        command_channel: MpscReceiver<MpvCommand>,
+        event_listeners: BroadcastSender<MpvEvent>,
+    ) -> Self {
+        Self {
+            mpv,
+            command_channel,
+            event_listeners,
+        }
+    }
+
+    pub async fn run(&mut self) -> anyhow::Result<()> {
+        loop {
+            tokio::select! {
+                Some(command) = self.command_channel.recv() => {
+                    self.mpv.run_command(command)?;
+                }
+                Ok(event) = async { self.mpv.event_listen() } => {
+                    self.event_listeners.send(event)?;
+                }
+            }
+        }
+    }
+}
+
+pub struct MpvConnectionArgs {
+    pub socket_path: String,
+    pub executable_path: Option<String>,
+    pub auto_start: bool,
+    pub force_auto_start: bool,
+}
+
+pub async fn connect_to_mpv(args: &MpvConnectionArgs) -> anyhow::Result<(Mpv, Option<Child>)> {
+    log::debug!("Connecting to mpv");
+
+    debug_assert!(
+        !args.force_auto_start || args.auto_start,
+        "force_auto_start requires auto_start"
+    );
+
+    let socket_path = Path::new(&args.socket_path);
+
+    if !socket_path.exists() {
+        log::debug!("Mpv socket not found at {}", &args.socket_path);
+        if !args.auto_start {
+            panic!("Mpv socket not found at {}", &args.socket_path);
+        }
+
+        log::debug!("Ensuring parent dir of mpv socket exists");
+        let parent_dir = Path::new(&args.socket_path)
+            .parent()
+            .context("Failed to get parent dir of mpv socket")?;
+
+        if !parent_dir.is_dir() {
+            create_dir_all(parent_dir).context("Failed to create parent dir of mpv socket")?;
+        }
+    } else {
+        log::debug!("Existing mpv socket found at {}", &args.socket_path);
+        if args.force_auto_start {
+            log::debug!("Removing mpv socket");
+            std::fs::remove_file(&args.socket_path)?;
+        }
+    }
+
+    let process_handle = if args.auto_start {
+        log::info!("Starting mpv with socket at {}", &args.socket_path);
+
+        // TODO: try to fetch mpv from PATH
+        Some(
+            Command::new(args.executable_path.as_deref().unwrap_or("mpv"))
+                .arg(format!("--input-ipc-server={}", &args.socket_path))
+                .arg("--idle")
+                .arg("--force-window")
+                // .arg("--fullscreen")
+                // .arg("--no-terminal")
+                // .arg("--load-unsafe-playlists")
+                .arg("--keep-open") // Keep last frame of video on end of video
+                .arg("--really-quiet")
+                .spawn()
+                .context("Failed to start mpv")?,
+        )
+    } else {
+        None
+    };
+
+    // Wait for mpv to create the socket
+    if tokio::time::timeout(tokio::time::Duration::from_millis(500), async {
+        while !&socket_path.exists() {
+            log::debug!("Waiting for mpv socket at {}", &args.socket_path);
+            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .is_err()
+    {
+        return Err(anyhow::anyhow!(
+            "Failed to connect to mpv socket: {}",
+            &args.socket_path
+        ));
+    }
+
+    Ok((
+        Mpv::connect(&args.socket_path).context(format!(
+            "Failed to connect to mpv socket: {}",
+            &args.socket_path
+        ))?,
+        process_handle,
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mpvipc_async::MpvCommand;
+    use tokio::sync::{broadcast, mpsc};
+
+    #[tokio::test]
+    async fn test_run() -> anyhow::Result<()> {
+        let (command_tx, command_rx) = mpsc::channel(1);
+        let (event_tx, _) = broadcast::channel(1);
+
+        let (mpv, _) = connect_to_mpv(&MpvConnectionArgs {
+            socket_path: "/tmp/mpv-test.sock".to_string(),
+            executable_path: None,
+            auto_start: true,
+            force_auto_start: true,
+        }).await?;
+
+        let mut broker = MpvBroker::new(mpv, command_rx, event_tx);
+        let broker_handle = tokio::spawn(async move {
+          broker.run().await.unwrap();
+        });
+
+        let _ = command_tx.send(MpvCommand::PlaylistClear).await;
+        let _ = broker_handle.await.unwrap();
+
+        Ok(())
+    }
+}