WIP: use new async mpvipc

src/api/base.rs

@@ -1,61 +1,57 @@
-use std::sync::Arc;
-
 use log::trace;
 use mpvipc::{
     Mpv, 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<()> {
+pub async fn loadfile(mpv: Mpv, path: &str) -> anyhow::Result<()> {
     trace!("api::loadfile({:?})", path);
-    mpv.lock().await.playlist_add(
+    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> {
+pub async fn play_get(mpv: Mpv) -> anyhow::Result<Value> {
     trace!("api::play_get()");
-    let paused: bool = mpv.lock().await.get_property("pause")?;
+    let paused: bool = mpv.get_property("pause").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<()> {
+pub async fn play_set(mpv: Mpv, should_play: bool) -> anyhow::Result<()> {
     trace!("api::play_set({:?})", should_play);
-    mpv.lock()
+    mpv.set_property("pause", !should_play)
         .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> {
+pub async fn volume_get(mpv: Mpv) -> anyhow::Result<Value> {
     trace!("api::volume_get()");
-    let volume: f64 = mpv.lock().await.get_property("volume")?;
+    let volume: f64 = mpv.get_property("volume").await?;
     Ok(json!(volume))
 }
 
 /// Set the player volume
-pub async fn volume_set(mpv: Arc<Mutex<Mpv>>, value: f64) -> anyhow::Result<()> {
+pub async fn volume_set(mpv: Mpv, value: f64) -> anyhow::Result<()> {
     trace!("api::volume_set({:?})", value);
-    mpv.lock()
+    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> {
+pub async fn time_get(mpv: 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 current: f64 = mpv.get_property("time-pos").await?;
+    let remaining: f64 = mpv.get_property("time-remaining").await?;
     let total = current + remaining;
 
     Ok(json!({
@@ -66,22 +62,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<()> {
+pub async fn time_set(mpv: Mpv, pos: Option<f64>, percent: Option<f64>) -> anyhow::Result<()> {
     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 +80,10 @@ pub async fn time_set(
 }
 
 /// Get the current playlist
-pub async fn playlist_get(mpv: Arc<Mutex<Mpv>>) -> anyhow::Result<Value> {
+pub async fn playlist_get(mpv: 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")?;
+    let playlist: mpvipc::Playlist = mpv.get_playlist().await?;
+    let is_playing: bool = mpv.get_property("pause").await?;
 
     let items: Vec<Value> = playlist
         .0
@@ -116,74 +106,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<()> {
+pub async fn playlist_next(mpv: Mpv) -> anyhow::Result<()> {
     trace!("api::playlist_next()");
-    mpv.lock().await.next().map_err(|e| e.into())
+    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<()> {
+pub async fn playlist_previous(mpv: Mpv) -> anyhow::Result<()> {
     trace!("api::playlist_previous()");
-    mpv.lock().await.prev().map_err(|e| e.into())
+    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<()> {
+pub async fn playlist_goto(mpv: Mpv, index: usize) -> anyhow::Result<()> {
     trace!("api::playlist_goto({:?})", index);
-    mpv.lock()
-        .await
-        .playlist_play_id(index)
-        .map_err(|e| e.into())
+    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<()> {
+pub async fn playlist_clear(mpv: Mpv) -> anyhow::Result<()> {
     trace!("api::playlist_clear()");
-    mpv.lock().await.playlist_clear().map_err(|e| e.into())
+    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<()> {
+pub async fn playlist_remove(mpv: Mpv, index: usize) -> anyhow::Result<()> {
     trace!("api::playlist_remove({:?})", index);
-    mpv.lock()
-        .await
-        .playlist_remove_id(index)
-        .map_err(|e| e.into())
+    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<()> {
+pub async fn playlist_move(mpv: 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())
+    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<()> {
+pub async fn shuffle(mpv: Mpv) -> anyhow::Result<()> {
     trace!("api::shuffle()");
-    mpv.lock().await.playlist_shuffle().map_err(|e| e.into())
+    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> {
+pub async fn playlist_get_looping(mpv: Mpv) -> anyhow::Result<Value> {
     trace!("api::playlist_get_looping()");
-    let loop_playlist = mpv.lock().await.get_property_string("loop-playlist")? == "inf";
+    let loop_playlist =
+        mpv.get_property_value("loop-playlist").await? == Value::String("inf".to_owned());
     Ok(json!(loop_playlist))
 }
 
-pub async fn playlist_set_looping(mpv: Arc<Mutex<Mpv>>, r#loop: bool) -> anyhow::Result<()> {
+pub async fn playlist_set_looping(mpv: Mpv, r#loop: bool) -> anyhow::Result<()> {
     trace!("api::playlist_set_looping({:?})", r#loop);
     if r#loop {
-        mpv.lock()
+        mpv.set_loop_playlist(Switch::On)
             .await
-            .set_loop_playlist(Switch::On)
             .map_err(|e| e.into())
     } else {
-        mpv.lock()
+        mpv.set_loop_playlist(Switch::Off)
             .await
-            .set_loop_playlist(Switch::Off)
             .map_err(|e| e.into())
     }
 }

src/api/rest_wrapper_v1.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use axum::{
+    debug_handler,
     extract::{Query, State},
     http::StatusCode,
     response::{IntoResponse, Response},
@@ -13,7 +14,7 @@ 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))
@@ -72,14 +73,12 @@ struct LoadFileArgs {
     path: String,
 }
 
-async fn loadfile(
-    State(mpv): State<Arc<Mutex<Mpv>>>,
-    Query(query): Query<LoadFileArgs>,
-) -> RestResponse {
+#[debug_handler]
+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 +87,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 +101,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 +115,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 +137,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 +149,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 +165,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 +173,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 +187,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

@@ -6,12 +6,8 @@ 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;
 
@@ -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,16 +142,17 @@ 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?);
+              mpv.disconnect().await?;
           }
           _ = tokio::signal::ctrl_c() => {
               log::info!("Received Ctrl-C, exiting");
+              mpv.disconnect().await?;
               proc.kill().await?;
           }
           result = async {
@@ -165,6 +162,7 @@ async fn main() -> anyhow::Result<()> {
               }
             } => {
               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?;
             }
         }
     }