WIP

Cargo.toml

@@ -10,10 +10,11 @@ readme = "README.md"
 
 [dependencies]
 anyhow = "1.0.82"
-axum = { version = "0.7.7", features = ["macros"] }
+axum = { version = "0.7.7", features = ["macros", "ws"] }
 clap = { version = "4.4.1", features = ["derive"] }
 clap-verbosity-flag = "2.2.2"
 env_logger = "0.10.0"
+futures = "0.3.31"
 log = "0.4.20"
 mpvipc-async = { git = "https://git.pvv.ntnu.no/Projects/mpvipc-async.git", rev = "v0.1.0" }
 sd-notify = "0.4.3"

src/api.rs

@@ -1,4 +1,6 @@
 mod base;
 mod rest_wrapper_v1;
+mod websocket_v1;
 
 pub use rest_wrapper_v1::{rest_api_docs, rest_api_routes};
+pub use websocket_v1::websocket_api;

src/api/websocket_v1.rs

@@ -0,0 +1,277 @@
+use std::net::SocketAddr;
+
+use anyhow::Context;
+use futures::{stream::FuturesUnordered, StreamExt};
+use serde::{Deserialize, Serialize};
+
+use axum::{
+    extract::{
+        ws::{Message, WebSocket},
+        ConnectInfo, State, WebSocketUpgrade,
+    },
+    response::IntoResponse,
+    routing::any,
+    Router,
+};
+use mpvipc_async::{
+    LoopProperty, Mpv, MpvExt, NumberChangeOptions, Playlist, PlaylistAddTypeOptions, SeekOptions,
+    Switch,
+};
+use serde_json::{json, Value};
+use tokio::select;
+
+pub fn websocket_api(mpv: Mpv) -> Router {
+    Router::new()
+        .route("/", any(websocket_handler))
+        .with_state(mpv)
+}
+
+async fn websocket_handler(
+    ws: WebSocketUpgrade,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    State(mpv): State<Mpv>,
+) -> impl IntoResponse {
+    let mpv = mpv.clone();
+    ws.on_upgrade(move |socket| handle_connection(socket, addr, mpv))
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct InitialState {
+    pub current_time: Option<f64>,
+    pub duration: f64,
+    pub volume: f64,
+    pub is_playing: bool,
+    pub is_looping: bool,
+    pub playlist: Playlist,
+    pub current_track: String,
+}
+
+async fn get_initial_state(mpv: &Mpv) -> InitialState {
+    let current_time = mpv.get_time_pos().await.unwrap_or(None);
+    let duration = mpv.get_duration().await.unwrap_or(0.0);
+    let volume = mpv.get_volume().await.unwrap_or(0.0);
+    let is_playing = mpv.is_playing().await.unwrap_or(false);
+    let is_looping =
+        mpv.playlist_is_looping().await.unwrap_or(LoopProperty::No) != LoopProperty::No;
+    // TODO: use default when new version is released
+    let playlist = mpv.get_playlist().await.unwrap_or(Playlist(vec![]));
+    let current_track = mpv.get_file_path().await.unwrap_or("".to_string());
+    InitialState {
+        current_time,
+        duration,
+        volume,
+        is_playing,
+        is_looping,
+        playlist,
+        current_track,
+    }
+}
+
+const DEFAULT_PROPERTY_SUBSCRIBES: [&str; 6] = [
+    "duration",
+    "loop-playlist",
+    "pause",
+    "playlist",
+    "time-pos",
+    "volume",
+];
+
+async fn setup_default_subscribes(mpv: &Mpv) -> anyhow::Result<()> {
+    let mut futures = FuturesUnordered::new();
+
+    futures.extend(
+        DEFAULT_PROPERTY_SUBSCRIBES
+            .iter()
+            .map(|property| mpv.observe_property(0, property)),
+    );
+
+    while let Some(result) = futures.next().await {
+        result?;
+    }
+
+    Ok(())
+}
+
+async fn handle_connection(mut socket: WebSocket, addr: SocketAddr, mpv: Mpv) {
+    // TODO: There is an asynchronous gap between gathering the initial state and subscribing to the properties
+    //       This could lead to missing events if they happen in that gap. Send initial state, but also ensure
+    //       that there is an additional "initial state" sent upon subscription to all properties to ensure that
+    //       the state is correct.
+    let initial_state = get_initial_state(&mpv).await;
+
+    let message = Message::Text(
+        json!({
+            "type": "initial_state",
+            "value": initial_state,
+        })
+        .to_string(),
+    );
+
+    socket.send(message).await.unwrap();
+
+    setup_default_subscribes(&mpv).await.unwrap();
+
+    let mut event_stream = mpv.get_event_stream().await;
+    let connection_loop = tokio::spawn(async move {
+        loop {
+            select! {
+              message = socket.recv() => {
+                  log::trace!("Received command from {:?}: {:?}", addr, message);
+
+                  let ws_message_content = message
+                  .ok_or(anyhow::anyhow!("Event stream ended for {:?}", addr))
+                  .and_then(|message| {
+                    match message {
+                      Ok(message) => Ok(message),
+                      err => Err(anyhow::anyhow!("Error reading message for {:?}: {:?}", addr, err)),
+                    }
+                  }).unwrap();
+
+                  if let Message::Close(_) = ws_message_content {
+                    log::trace!("Closing connection for {:?}", addr);
+                    return Ok(());
+                  }
+
+                  if let Message::Ping(xs) = ws_message_content {
+                    log::trace!("Ponging {:?} with {:?}", addr, xs);
+                    socket.send(Message::Pong(xs)).await.unwrap();
+                    continue;
+                  }
+
+                  let message_content = match ws_message_content {
+                      Message::Text(text) => text,
+                      m => return Err(anyhow::anyhow!("Unexpected message type: {:?}", m)),
+                  };
+
+                  let message_json = match serde_json::from_str::<Value>(&message_content) {
+                      Ok(json) => json,
+                      Err(e) => return Err(anyhow::anyhow!("Error parsing message from {:?}: {:?}", addr, e)),
+                  };
+
+                  // TODO: handle errors
+                  if let Ok(Some(response)) = handle_message(message_json, mpv.clone()).await {
+                      let message = Message::Text(json!({
+                        "type": "response",
+                        "value": response,
+                      }).to_string());
+                      socket.send(message).await.unwrap();
+                  }
+              }
+              event = event_stream.next() => {
+                match event {
+                  Some(Ok(event)) => {
+                    log::trace!("Sending event to {:?}: {:?}", addr, event);
+                    let message = Message::Text(json!({
+                      "type": "event",
+                      "value": event,
+                    }).to_string());
+                    socket.send(message).await.unwrap();
+                  }
+                  Some(Err(e)) => {
+                    log::error!("Error reading event stream for {:?}: {:?}", addr, e);
+                    anyhow::bail!("Error reading event stream for {:?}: {:?}", addr, e);
+                  }
+                  None => {
+                    log::trace!("Event stream ended for {:?}", addr);
+                    return Ok(());
+                  }
+                }
+              }
+            }
+        }
+    });
+
+    connection_loop.await.unwrap().unwrap()
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum WSCommand {
+    Subscribe { property: String },
+    UnsubscribeAll,
+
+    Load { url: String },
+    TogglePlayback,
+    Volume { volume: f64 },
+    Time { time: f64 },
+    PlaylistNext,
+    PlaylistPrevious,
+    PlaylistGoto { position: usize },
+    PlaylistClear,
+    PlaylistRemove { position: usize },
+    PlaylistMove { from: usize, to: usize },
+    Shuffle,
+    SetLooping { value: bool },
+}
+
+async fn handle_message(message: Value, mpv: Mpv) -> anyhow::Result<Option<Value>> {
+    let command =
+        serde_json::from_value::<WSCommand>(message).context("Failed to parse message")?;
+
+    match command {
+        WSCommand::Subscribe { property } => {
+            // TODO: get an id provisioned by the id pool
+            mpv.observe_property(0, &property).await?;
+            Ok(None)
+        }
+        WSCommand::UnsubscribeAll => {
+            mpv.unobserve_property(0).await?;
+            Ok(None)
+        }
+        WSCommand::Load { url } => {
+            mpv.playlist_add(
+                &url,
+                PlaylistAddTypeOptions::File,
+                mpvipc_async::PlaylistAddOptions::Append,
+            )
+            .await?;
+            Ok(None)
+        }
+        WSCommand::TogglePlayback => {
+            mpv.set_playback(mpvipc_async::Switch::Toggle).await?;
+            Ok(None)
+        }
+        WSCommand::Volume { volume } => {
+            mpv.set_volume(volume, NumberChangeOptions::Absolute)
+                .await?;
+            Ok(None)
+        }
+        WSCommand::Time { time } => {
+            mpv.seek(time, SeekOptions::Absolute).await?;
+            Ok(None)
+        }
+        WSCommand::PlaylistNext => {
+            mpv.next().await?;
+            Ok(None)
+        }
+        WSCommand::PlaylistPrevious => {
+            mpv.prev().await?;
+            Ok(None)
+        }
+        WSCommand::PlaylistGoto { position } => {
+            mpv.playlist_play_id(position).await?;
+            Ok(None)
+        }
+        WSCommand::PlaylistClear => {
+            mpv.playlist_clear().await?;
+            Ok(None)
+        }
+        WSCommand::PlaylistRemove { position } => {
+            mpv.playlist_remove_id(position).await?;
+            Ok(None)
+        }
+        WSCommand::PlaylistMove { from, to } => {
+            mpv.playlist_move_id(from, to).await?;
+            Ok(None)
+        }
+        WSCommand::Shuffle => {
+            mpv.playlist_shuffle().await?;
+            Ok(None)
+        }
+        WSCommand::SetLooping { value } => {
+            mpv.set_loop_playlist(if value { Switch::On } else { Switch::Off })
+                .await?;
+            Ok(None)
+        }
+    }
+}

src/main.rs

@@ -162,7 +162,9 @@ async fn main() -> anyhow::Result<()> {
 
     let app = Router::new()
         .nest("/api", api::rest_api_routes(mpv.clone()))
-        .merge(api::rest_api_docs(mpv.clone()));
+        .nest("/ws", api::websocket_api(mpv.clone()))
+        .merge(api::rest_api_docs(mpv.clone()))
+        .into_make_service_with_connect_info::<SocketAddr>();
 
     let listener = match tokio::net::TcpListener::bind(&socket_addr)
         .await
@@ -199,7 +201,7 @@ async fn main() -> anyhow::Result<()> {
                 log::info!("Received Ctrl-C, exiting");
                 shutdown(mpv, Some(proc)).await;
             }
-            result = axum::serve(listener, app.into_make_service()) => {
+            result = axum::serve(listener, app) => {
               log::info!("API server exited");
               shutdown(mpv, Some(proc)).await;
               result?;
@@ -211,7 +213,7 @@ async fn main() -> anyhow::Result<()> {
                 log::info!("Received Ctrl-C, exiting");
                 shutdown(mpv.clone(), None).await;
             }
-            result = axum::serve(listener, app.into_make_service()) => {
+            result = axum::serve(listener, app) => {
               log::info!("API server exited");
               shutdown(mpv.clone(), None).await;
               result?;