//! The cbd-tui client as a library: the server-facing orchestration loop //! and the terminal UI loop, exposed as [`run`] so both the standalone //! `cbd-tui` binary and the bundled `cbd` binary can host them //! (architecture/cbd-bundle.md D1). Tracing setup stays with the //! binaries — where logs go is a hosting decision. pub mod app; pub mod config; pub mod rpc; #[cfg(feature = "mpris")] pub mod mpris; /// Built without the `mpris` feature: the same shape, doing nothing, so the /// orchestrator needs no `cfg` of its own (architecture/build-features.md D1). #[cfg(not(feature = "mpris"))] pub mod mpris { use crabidy_core::proto::crabidy::{ get_update_stream_response::Update as StreamUpdate, InitResponse, }; use flume::Sender; use crate::app::MessageFromUi; /// There is no player to feed. The orchestrator holds an `Option` of this /// and never gets a `Some`. #[derive(Debug)] pub struct Feed; impl Feed { pub fn publish(&self, _update: &StreamUpdate) {} pub fn publish_init(&self, _init: &InitResponse) {} } pub async fn start(_commands: Sender) -> Option { None } } use std::{ error::Error, io, time::{Duration, Instant}, }; use crabidy_core::proto::crabidy::{get_update_stream_response::Update as StreamUpdate, PlayState}; use crossterm::{ event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyEventKind}, execute, terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, }; use flume::{Receiver, Sender}; use ratatui::{backend::CrosstermBackend, Terminal}; use tokio::select; use tokio_stream::StreamExt; use app::{bindings, App, DispatchResult, MessageFromUi, MessageToUi}; use config::Config; use rpc::RpcClient; use tracing::{error, info, warn}; /// Runs the client: the rpc orchestration loop on the runtime, the /// blocking terminal UI on its own thread. Returns when the user quits /// the UI. pub async fn run(config: &'static Config) -> Result<(), Box> { let (ui_tx, rx): (Sender, Receiver) = flume::unbounded(); let (tx, ui_rx): (Sender, Receiver) = flume::unbounded(); // The MPRIS player commands the server through the same channel the // keybindings use, so it is just another producer of `MessageFromUi` // (architecture/mpris.md D1). let commands = ui_tx.clone(); // FIXME: unwrap tokio::spawn(async move { orchestrate(config, (tx, rx), commands).await.unwrap() }); let spectrum_enabled = config.server.spectrum; // Resolved here rather than in the UI thread so a bad config string is // reported on stderr before the alternate screen swallows it. let spectrum_style = config::spectrum_style(config); tokio::task::spawn_blocking(move || { run_ui(ui_tx, ui_rx, spectrum_enabled, spectrum_style); }) .await?; Ok(()) } async fn orchestrate( config: &'static Config, (tx, rx): (Sender, Receiver), commands: Sender, ) -> Result<(), Box> { info!(address = config.server.address, "connecting to server"); let mut rpc_client = rpc::RpcClient::connect(&config.server).await?; if let Some(root_node) = rpc_client.get_library_node(crabidy_core::ROOT_PATH).await? { tx.send(MessageToUi::ReplaceLibraryNode(root_node.clone()))?; } // A desktop that cannot host the player (no session bus) leaves this // `None` and changes nothing else (architecture/mpris.md D13). let mpris = mpris::start(commands).await; let init_data = rpc_client.init().await?; info!("received initial state from server"); if let Some(mpris) = &mpris { mpris.publish_init(&init_data); } tx.send_async(MessageToUi::Init(init_data)).await?; loop { if let Err(err) = poll(&mut rpc_client, &rx, &tx, &mpris).await { error!("request to server failed: {err}"); } } } async fn poll( rpc_client: &mut RpcClient, rx: &Receiver, tx: &Sender, mpris: &Option, ) -> Result<(), Box> { select! { Ok(msg) = &mut rx.recv_async() => { match msg { MessageFromUi::GetLibraryNode(path) => { if let Some(node) = rpc_client.get_library_node(&path).await? { let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); } }, MessageFromUi::CreateNode { parent_path, title } => { // Navigates the library into the created node on // success; on failure the library stays where it is. match rpc_client.create_library_node(&parent_path, &title).await { Ok(node) => { let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); } Err(err) => { error!(parent_path, title, "failed to create node: {err}"); } } }, MessageFromUi::RenameNode { path, new_title } => { // Navigates the library into the renamed node on success; // on failure the library stays where it is. match rpc_client.rename_library_node(&path, &new_title).await { Ok(node) => { let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); } Err(err) => { error!(path, new_title, "failed to rename node: {err}"); } } }, MessageFromUi::DeleteNode { path } => { // Shows the refreshed parent listing on success; on // failure the library stays where it is. match rpc_client.delete_library_node(&path).await { Ok(parent) => { let _ = tx.send(MessageToUi::ReplaceLibraryNode(parent.clone())); } Err(err) => { error!(path, "failed to delete node: {err}"); } } }, MessageFromUi::AppendTracks(uuids) => { rpc_client.append_tracks(uuids).await? } MessageFromUi::QueueTracks(uuids) => { rpc_client.queue_tracks(uuids).await? } MessageFromUi::InsertTracks(uuids, pos) => { rpc_client.insert_tracks(uuids, pos).await? } MessageFromUi::RemoveTracks(positions) => { rpc_client.remove_tracks(positions).await? } MessageFromUi::ReplaceQueue(uuids) => { rpc_client.replace_queue(uuids).await? } MessageFromUi::NextTrack => { rpc_client.next_track().await? } MessageFromUi::PrevTrack => { rpc_client.prev_track().await? } MessageFromUi::RestartTrack => { rpc_client.restart_track().await? } MessageFromUi::Seek(delta_millis) => { rpc_client.seek(delta_millis).await? } MessageFromUi::SetCurrentTrack(pos) => { rpc_client.set_current_track(pos).await? } MessageFromUi::TogglePlay => { rpc_client.toggle_play().await? } MessageFromUi::Stop => { rpc_client.stop().await? } MessageFromUi::ChangeVolume(delta) => { rpc_client.change_volume(delta).await? } MessageFromUi::ToggleMute => { rpc_client.toggle_mute().await? } MessageFromUi::ToggleShuffle => { rpc_client.toggle_shuffle().await? } MessageFromUi::ToggleRepeat => { rpc_client.toggle_repeat().await? } MessageFromUi::ClearQueue(exclude_current) => { rpc_client.clear_queue(exclude_current).await? } MessageFromUi::DedupQueue => { // The count is the whole point (architecture/queue-order.md // D2); a failure is logged and the queue simply stays as it // is — it must not tear down the poll loop. match rpc_client.dedup_queue().await { Ok(removed) => { let _ = tx.send(MessageToUi::QueueDeduped { removed }); } Err(err) => error!("failed to dedup the queue: {err}"), } } MessageFromUi::SortQueue { sort, descending } => { if let Err(err) = rpc_client.sort_queue(sort, descending).await { error!(?sort, descending, "failed to sort the queue: {err}"); } } MessageFromUi::SaveQueue(name) => { // A rejected save (bad name, empty queue) must not tear // down the poll loop; the server logs the cause. if let Err(err) = rpc_client.save_queue(name.clone()).await { error!(name, "failed to save queue: {err}"); } } MessageFromUi::CaptureNode { path, name, download } => { // A rejected capture (bad name, over-cap subtree, failed // download) must not tear down the poll loop either. if let Err(err) = rpc_client .capture_library_node(path.clone(), name.clone(), download) .await { error!(path, name, download, "failed to capture subtree: {err}"); } } } } Some(resp) = rpc_client.update_stream.next() => { match resp { Ok(resp) => { if let Some(update) = resp.update { // The UI thread and the MPRIS player are peers on one // stream; neither learns anything the other does not. if let Some(mpris) = mpris { mpris.publish(&update); } tx.send_async(MessageToUi::Update(update)).await?; } } Err(err) => { warn!("update stream broke, reconnecting: {err}"); rpc_client.reconnect_update_stream().await; info!("update stream reconnected"); } } } } Ok(()) } fn run_ui( tx: Sender, rx: Receiver, spectrum_enabled: bool, spectrum_style: app::SpectrumStyle, ) { // setup terminal enable_raw_mode().unwrap(); let mut stdout = io::stdout(); execute!(stdout, EnterAlternateScreen, EnableMouseCapture).unwrap(); let backend = CrosstermBackend::new(stdout); let mut terminal = Terminal::new(backend).unwrap(); // create app and run it let mut app = App::new(tx); app.now_playing.set_spectrum_enabled(spectrum_enabled); app.now_playing.set_spectrum_style(spectrum_style); let tick_rate = Duration::from_millis(100); let mut last_tick = Instant::now(); loop { for message in rx.try_iter() { match message { MessageToUi::ReplaceLibraryNode(node) => { app.library.update(node); } MessageToUi::Init(init_data) => { if let Some(queue) = init_data.queue { app.queue.update_queue(queue); } if let Some(track) = init_data.queue_track { app.now_playing.update_track(track.track); app.queue.update_position(track.queue_position as usize); } if let Ok(ps) = PlayState::try_from(init_data.play_state) { app.now_playing.update_play_state(ps); } if let Some(mods) = init_data.mods { app.now_playing.update_modifiers(&mods); } // Both were dropped here, so the pane showed a default // volume and an unmuted server until the user first // touched either — a display that starts out wrong. app.now_playing.update_volume(init_data.volume); app.now_playing.update_mute(init_data.mute); } MessageToUi::Update(update) => match update { StreamUpdate::Queue(queue) => { app.queue.update_queue(queue); } StreamUpdate::QueueTrack(track) => { app.now_playing.update_track(track.track); app.queue.update_position(track.queue_position as usize); } StreamUpdate::Position(pos) => app.now_playing.update_position(pos), StreamUpdate::PlayState(play_state) => { if let Ok(ps) = PlayState::try_from(play_state) { app.now_playing.update_play_state(ps); } } StreamUpdate::Mods(mods) => { app.now_playing.update_modifiers(&mods); } StreamUpdate::Mute(muted) => app.now_playing.update_mute(muted), StreamUpdate::Volume(volume) => app.now_playing.update_volume(volume), StreamUpdate::CaptureProgress(progress) => { app.captures.apply(progress); } StreamUpdate::Spectrum(frame) => { app.now_playing.update_spectrum(frame.bins); } }, MessageToUi::QueueDeduped { removed } => { app.queue.show_dedup_result(removed); } } } if let Err(err) = terminal.draw(|f| app.render(f)) { error!("failed to draw frame: {err}"); break; } let timeout = tick_rate .checked_sub(last_tick.elapsed()) .unwrap_or_else(|| Duration::from_secs(0)); if event::poll(timeout).unwrap() { if let Event::Key(key) = event::read().unwrap() { if key.kind == KeyEventKind::Press { // The overlays are strictly modal: while one is open, // keys answer it and the bindings table (including // quit) is unreachable. if app.search.is_some() { app.handle_search_key(key); } else if app.input.is_some() { app.handle_input_key(key); } else if app.sort_menu { app.handle_sort_key(key); } else if let Some(action) = bindings::lookup(app.focus, app.show_help, key) { if app.dispatch(action) == DispatchResult::Quit { break; } } } } } if last_tick.elapsed() >= tick_rate { last_tick = Instant::now(); } } // restore terminal disable_raw_mode().unwrap(); execute!( terminal.backend_mut(), LeaveAlternateScreen, DisableMouseCapture ) .unwrap(); terminal.show_cursor().unwrap(); }