From 0a5e6b738ed37878fe23d54a2f90e003ac9bef43 Mon Sep 17 00:00:00 2001 From: h7x4 Date: Tue, 6 May 2025 21:09:44 +0200 Subject: [PATCH] Add connection counter to status, print status in non-systemd mode --- src/api/websocket_v1.rs | 52 +++++++++++-- src/main.rs | 129 +++++++++++++++++++++++---------- src/util.rs | 2 + src/util/connection_counter.rs | 25 +++++++ 4 files changed, 163 insertions(+), 45 deletions(-) create mode 100644 src/util/connection_counter.rs diff --git a/src/api/websocket_v1.rs b/src/api/websocket_v1.rs index 8436e4f..eca1103 100644 --- a/src/api/websocket_v1.rs +++ b/src/api/websocket_v1.rs @@ -21,18 +21,30 @@ use mpvipc_async::{ Switch, }; use serde_json::{json, Value}; -use tokio::{select, sync::watch}; +use tokio::{ + select, + sync::{mpsc, watch}, +}; -use crate::util::IdPool; +use crate::util::{ConnectionEvent, IdPool}; #[derive(Debug, Clone)] struct WebsocketState { mpv: Mpv, id_pool: Arc>, + connection_counter_tx: mpsc::Sender, } -pub fn websocket_api(mpv: Mpv, id_pool: Arc>) -> Router { - let state = WebsocketState { mpv, id_pool }; +pub fn websocket_api( + mpv: Mpv, + id_pool: Arc>, + connection_counter_tx: mpsc::Sender, +) -> Router { + let state = WebsocketState { + mpv, + id_pool, + connection_counter_tx, + }; Router::new() .route("/", any(websocket_handler)) .with_state(state) @@ -41,7 +53,11 @@ pub fn websocket_api(mpv: Mpv, id_pool: Arc>) -> Router { async fn websocket_handler( ws: WebSocketUpgrade, ConnectInfo(addr): ConnectInfo, - State(WebsocketState { mpv, id_pool }): State, + State(WebsocketState { + mpv, + id_pool, + connection_counter_tx, + }): State, ) -> impl IntoResponse { let mpv = mpv.clone(); let id = match id_pool.lock().unwrap().request_id() { @@ -52,7 +68,9 @@ async fn websocket_handler( } }; - ws.on_upgrade(move |socket| handle_connection(socket, addr, mpv, id, id_pool)) + ws.on_upgrade(move |socket| { + handle_connection(socket, addr, mpv, id, id_pool, connection_counter_tx) + }) } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -174,7 +192,17 @@ async fn handle_connection( mpv: Mpv, channel_id: u64, id_pool: Arc>, + connection_counter_tx: mpsc::Sender, ) { + match connection_counter_tx.send(ConnectionEvent::Connected).await { + Ok(()) => { + log::trace!("Connection count updated for {:?}", addr); + } + Err(e) => { + log::error!("Error updating connection count for {:?}: {:?}", addr, e); + } + } + // 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 @@ -236,6 +264,18 @@ async fn handle_connection( log::error!("Error releasing id {} for {:?}: {:?}", channel_id, addr, e); } } + + match connection_counter_tx + .send(ConnectionEvent::Disconnected) + .await + { + Ok(()) => { + log::trace!("Connection count updated for {:?}", addr); + } + Err(e) => { + log::error!("Error updating connection count for {:?}: {:?}", addr, e); + } + } } async fn connection_loop( diff --git a/src/main.rs b/src/main.rs index bd9d29a..2be6267 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,8 +11,8 @@ use std::{ }; use systemd_journal_logger::JournalLog; use tempfile::NamedTempFile; -use tokio::task::JoinHandle; -use util::IdPool; +use tokio::{sync::mpsc, task::JoinHandle}; +use util::{ConnectionEvent, IdPool}; mod api; mod mpv_setup; @@ -94,23 +94,37 @@ async fn setup_systemd_watchdog_thread() -> anyhow::Result<()> { Ok(()) } -fn systemd_update_play_status(playing: bool, current_song: &Option) { - sd_notify::notify( - false, - &[sd_notify::NotifyState::Status(&format!( - "{} {:?}", - if playing { "[PLAY]" } else { "[STOP]" }, - if let Some(song) = current_song { - song - } else { - "" - } - ))], - ) - .unwrap_or_else(|e| log::warn!("Failed to update systemd status with current song: {}", e)); +fn send_play_status( + systemd: bool, + playing: bool, + current_song: &Option, + connection_count: u64, +) { + let status = &format!( + "[CONN: {}] {} {:?}", + connection_count, + if playing { "[PLAY]" } else { "[STOP]" }, + if let Some(song) = current_song { + song + } else { + "" + } + ); + + if systemd { + sd_notify::notify(false, &[sd_notify::NotifyState::Status(status)]).unwrap_or_else(|e| { + log::warn!("Failed to update systemd status with current song: {}", e) + }); + } else { + log::info!("{}", status); + } } -async fn setup_systemd_notifier(mpv: Mpv) -> anyhow::Result> { +async fn start_status_notifier_thread( + systemd: bool, + mpv: Mpv, + mut connection_counter_rx: mpsc::Receiver, +) -> anyhow::Result> { let handle = tokio::spawn(async move { log::debug!("Starting systemd notifier thread"); let mut event_stream = mpv.get_event_stream().await; @@ -120,30 +134,53 @@ async fn setup_systemd_notifier(mpv: Mpv) -> anyhow::Result> { let mut current_song: Option = mpv.get_property("media-title").await.unwrap(); let mut playing = !mpv.get_property("pause").await.unwrap().unwrap_or(false); + let mut connection_count = 0; - systemd_update_play_status(playing, ¤t_song); + send_play_status(systemd, playing, ¤t_song, connection_count); loop { - if let Some(Ok(Event::PropertyChange { name, data, .. })) = event_stream.next().await { - match (name.as_str(), data) { - ("media-title", Some(MpvDataType::String(s))) => { - current_song = Some(s); - } - ("media-title", None) => { - current_song = None; - } - ("pause", Some(MpvDataType::Bool(b))) => { - playing = !b; - } - (event_name, _) => { - log::trace!( - "Received unexpected property change on systemd notifier thread: {}", - event_name - ); + tokio::select! { + Some(Ok(Event::PropertyChange { name, data, .. })) = event_stream.next() => { + match (name.as_str(), data) { + ("media-title", Some(MpvDataType::String(s))) => { + current_song = Some(s); + } + ("media-title", None) => { + current_song = None; + } + ("pause", Some(MpvDataType::Bool(b))) => { + playing = !b; + } + (event_name, _) => { + log::trace!( + "Received unexpected property change on systemd notifier thread: {}", + event_name + ); + } } + + send_play_status(systemd, playing, ¤t_song, connection_count) } - systemd_update_play_status(playing, ¤t_song) + Some(connection_counter_update) = connection_counter_rx.recv() => { + log::trace!("Received connection counter update: {}", connection_counter_update); + + match connection_count.checked_add_signed(connection_counter_update.to_i8().into()) { + Some(new_count) => connection_count = new_count, + None => { + log::warn!("Invalid connection count: trying to add {} to {}", connection_counter_update.to_i8(), connection_count); + log::warn!("Resetting connection count to 0"); + connection_count = 0; + } + } + + match connection_count { + 0 => log::debug!("No connections"), + _ => log::debug!("Connection count: {}", connection_count), + } + + send_play_status(systemd, playing, ¤t_song, connection_count); + } } } }); @@ -206,9 +243,10 @@ async fn main() -> anyhow::Result<()> { .await .context("Failed to connect to mpv")?; - if systemd_mode { - setup_systemd_notifier(mpv.clone()).await?; - } + let (connection_counter_tx, connection_counter_rx) = mpsc::channel(10); + + let status_notifier_thread_handle = + start_status_notifier_thread(systemd_mode, mpv.clone(), connection_counter_rx).await?; if let Err(e) = show_grzegorz_image(mpv.clone()).await { log::warn!("Could not show Grzegorz image: {}", e); @@ -232,7 +270,10 @@ async fn main() -> anyhow::Result<()> { let app = Router::new() .nest("/api", api::rest_api_routes(mpv.clone())) - .nest("/ws", api::websocket_api(mpv.clone(), id_pool.clone())) + .nest( + "/ws", + api::websocket_api(mpv.clone(), id_pool.clone(), connection_counter_tx.clone()), + ) .merge(api::rest_api_docs(mpv.clone())) .into_make_service_with_connect_info::(); @@ -276,6 +317,11 @@ async fn main() -> anyhow::Result<()> { shutdown(mpv, Some(proc)).await; result?; } + result = status_notifier_thread_handle => { + log::info!("Status notifier thread exited unexpectedly, shutting dow"); + shutdown(mpv, Some(proc)).await; + result?; + } } } else { tokio::select! { @@ -288,6 +334,11 @@ async fn main() -> anyhow::Result<()> { shutdown(mpv.clone(), None).await; result?; } + result = status_notifier_thread_handle => { + log::info!("Status notifier thread exited unexpectedly, shutting down"); + shutdown(mpv.clone(), None).await; + result?; + } } } diff --git a/src/util.rs b/src/util.rs index 451206e..4b7c761 100644 --- a/src/util.rs +++ b/src/util.rs @@ -1,3 +1,5 @@ +mod connection_counter; mod id_pool; +pub use connection_counter::ConnectionEvent; pub use id_pool::IdPool; diff --git a/src/util/connection_counter.rs b/src/util/connection_counter.rs new file mode 100644 index 0000000..54e90a5 --- /dev/null +++ b/src/util/connection_counter.rs @@ -0,0 +1,25 @@ +use std::fmt; + +#[derive(Debug, Clone, Copy)] +pub enum ConnectionEvent { + Connected, + Disconnected, +} + +impl ConnectionEvent { + pub fn to_i8(self) -> i8 { + match self { + ConnectionEvent::Connected => 1, + ConnectionEvent::Disconnected => -1, + } + } +} + +impl fmt::Display for ConnectionEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ConnectionEvent::Connected => write!(f, "Connected"), + ConnectionEvent::Disconnected => write!(f, "Disconnected"), + } + } +}