selene_daemon/

daemon.rs

use std::{
    io::{self, Read, Write},
    sync::mpsc::Sender,
};

use selene_core::library::collection::Collectable;

use blake3::Hash;

use crate::{
    IpcCommand, IpcHandleError, IpcRequest, IpcTx, PacketError, PlayerEvent,
    player::{PlayerQueryFlags, QueryResult},
    playlist::{LoopMode, PlaybackStatus, ShuffleMode},
};

// Interface stuff
/// Daemon connection abstraction
///
/// Allow clients to connect to selene-daemon using one of the handles compatible with their system. Defines wrapper functions around all IPC calls to allow for easy get/set calls
pub struct SeleneDaemon {
    handle_tx: Sender<IpcRequest>,
}

// Daemon impls
impl SeleneDaemon {
    /// Create a new connection to the daemon
    ///
    /// # Errors
    ///
    /// This function will error if:
    /// - There is no [`DaemonHandle`] configured for the current system
    /// - The connection could not be established (May be because the daemon is not running)
    pub fn connect<F>(callback: Option<F>) -> Result<SeleneDaemon, IpcHandleError>
    where
        Self: Sized,
        F: FnMut(PlayerEvent) + Send + Sync + 'static,
    {
        #[cfg(unix)]
        {
            Ok(Self {
                handle_tx: unix_socket_handle::UnixSocketHandle::connect(callback)?,
            })
        }

        #[cfg(not(any(unix)))]
        {
            Err(IpcHandleError::UnsupportedPlatform)
        }
    }
}

impl SeleneDaemon {
    // Generic
    pub fn ipc_generic(&self, command: IpcCommand) -> Result<(), PacketError> {
        self.handle_tx.no_response(command)
    }

    pub fn ipc_flush(&self) -> Result<(), PacketError> {
        self.handle_tx.request(IpcCommand::Flush)
    }

    pub fn ipc_disconnect(&self) {
        self.handle_tx.action(IpcCommand::Disconnect);
    }

    pub fn ipc_reload_config(&self) -> Result<Result<(), String>, PacketError> {
        self.handle_tx.request(IpcCommand::ReloadConfig)
    }

    // Playback
    pub fn ipc_play(&self, collectable: Collectable) {
        self.handle_tx.action(IpcCommand::Play { collectable });
    }

    pub fn ipc_stop(&self) {
        self.handle_tx.action(IpcCommand::Stop);
    }

    pub fn ipc_set_is_playing(&self, is_playing: bool) {
        self.handle_tx
            .action(IpcCommand::SetIsPlaying { is_playing });
    }

    pub fn ipc_toggle_is_playing(&self) -> Result<PlaybackStatus, PacketError> {
        self.handle_tx.request(IpcCommand::TogglePlaying)
    }

    pub fn ipc_seek(&self, time: f64, increment: bool) -> Result<Option<f64>, PacketError> {
        self.handle_tx.request(IpcCommand::Seek {
            seconds: time,
            increment,
        })
    }

    pub fn ipc_set_volume(&self, volume: f32, increment: bool) -> Result<f32, PacketError> {
        self.handle_tx
            .request(IpcCommand::SetVolume { volume, increment })
    }

    // Queue
    pub fn ipc_queue_set(
        &self,
        tracks: Vec<Collectable>,
        expected_state: Hash,
    ) -> Result<bool, PacketError> {
        self.handle_tx.request(IpcCommand::QueueSet {
            tracks,
            expected_state,
        })
    }

    pub fn ipc_queue_extend(&self, tracks: Vec<Collectable>) {
        self.handle_tx.action(IpcCommand::QueueExtend(tracks));
    }

    pub fn ipc_queue_shuffle(&self) {
        self.handle_tx.action(IpcCommand::QueueShuffle);
    }

    pub fn ipc_queue_clear(&self) {
        self.handle_tx.action(IpcCommand::QueueClear);
    }

    // Playlist
    pub fn ipc_playlist_set(
        &self,
        collectables: Vec<Collectable>,
        expected_state: Hash,
    ) -> Result<bool, PacketError> {
        self.handle_tx.request(IpcCommand::PlaylistSet {
            collectables,
            expected_state,
        })
    }

    pub fn ipc_playlist_extend(&self, collectables: Vec<Collectable>) {
        self.handle_tx
            .action(IpcCommand::PlaylistExtend(collectables));
    }

    pub fn ipc_playlist_clear(&self) {
        self.handle_tx.action(IpcCommand::PlaylistClear);
    }

    pub fn ipc_playlist_set_shuffle_mode(&self, shuffle_mode: ShuffleMode) {
        self.handle_tx
            .action(IpcCommand::PlaylistSetShuffleMode { shuffle_mode });
    }

    pub fn ipc_playlist_set_loop_mode(&self, loop_mode: LoopMode) {
        self.handle_tx
            .action(IpcCommand::PlaylistSetLoopMode { loop_mode });
    }

    // Tracklist
    pub fn ipc_tracklist_rebuild(&self) {
        self.handle_tx.action(IpcCommand::TracklistRebuild);
    }

    pub fn ipc_tracklist_seek(&self, index: isize, increment: bool) -> Result<usize, PacketError> {
        self.handle_tx
            .request(IpcCommand::TracklistSeek { index, increment })
    }

    // Queries
    pub fn ipc_query(&self, flags: PlayerQueryFlags) -> Result<QueryResult, PacketError> {
        self.handle_tx.request(IpcCommand::GetState { flags })
    }
}

impl Drop for SeleneDaemon {
    fn drop(&mut self) {
        let () = self.handle_tx.disconnect();
    }
}

/// Daemon connection abstraction. Clients only need to be concerned with sending an IPC, and getting a response back, not how they are connecting
pub trait SeleneDaemonHandle: Send {
    /// Connects to the daemon and runs the handle in its own thread
    fn connect<F>(callback: Option<F>) -> Result<Sender<IpcRequest>, IpcHandleError>
    where
        Self: Sized,
        F: FnMut(PlayerEvent) + Send + Sync + 'static;

    /// Attempts to reconnect to the daemon
    fn reconnect(&mut self) -> bool;

    /// Main handle logic. This should run forever and in its own thread
    fn run<F>(self, callback: Option<F>)
    where
        F: FnMut(PlayerEvent) + Send + Sync + 'static;
}

/// Assumes the next two byte groups are \[len\]\[data\] and that the packet type has already been read
///
/// Returns just the packet data from the buffer
///
/// # Errors
///
/// Errors if [`read_exact()`] fails
fn read_packet_data(stream: &mut impl Read) -> io::Result<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    stream.read_exact(&mut len_buf)?;
    let len = u32::from_be_bytes(len_buf) as usize;

    let mut data = vec![0u8; len];
    stream.read_exact(&mut data)?;

    Ok(data)
}

fn send_command(
    stream: &mut impl Write,
    command: &IpcCommand,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut buf = Vec::new();
    ciborium::into_writer(&command, &mut buf).expect("Serialization should not fail.");
    stream.write_all(&(buf.len() as u32).to_be_bytes())?;
    stream.write_all(&buf)?;
    Ok(())
}

#[cfg(unix)]
/// `DaemonHandle` implementation for unix using unix sockets
pub mod unix_socket_handle;