selene_daemon/

client.rs

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

pub type CallbackFn = Box<dyn FnOnce(Result<&[u8], PacketError>) + Send>;

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

mod ipc;
pub use ipc::*;
use lunar_lib::config::ConfigError;
use thiserror::Error;

use crate::{player::PlayerEvent, wait};

#[derive(Debug, Error)]
pub enum IpcHandleError {
    #[error("Failed to connect: {0}")]
    FailedToConnect(ConnectErrorKind),

    #[error("The handling thread cannot be communicated with")]
    HandleDied,

    #[error("The current platform is not supported")]
    UnsupportedPlatform,
}

impl From<ConfigError> for IpcHandleError {
    fn from(value: ConfigError) -> Self {
        Self::FailedToConnect(ConnectErrorKind::FailedToLoadConfig(value.to_string()))
    }
}

#[derive(Debug)]
pub enum ConnectErrorKind {
    DaemonNotRunning,
    ConnectionRefused,
    FailedToLoadConfig(String),
    Other(io::Error),
}

impl Display for ConnectErrorKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ConnectErrorKind::DaemonNotRunning => f.write_str("The daemon is not running"),
            ConnectErrorKind::ConnectionRefused => {
                f.write_str("The daemon listener thread halted and must be restarted")
            }
            ConnectErrorKind::FailedToLoadConfig(string) => string.fmt(f),
            ConnectErrorKind::Other(error) => error.fmt(f),
        }
    }
}

// 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 SeleneClient {
    handle_tx: Sender<IpcRequest>,
}

// Daemon impls
impl SeleneClient {
    /// 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>(event_callback: F) -> Result<SeleneClient, IpcHandleError>
    where
        Self: Sized,
        F: FnMut(PlayerEvent) + Send + Sync + 'static,
    {
        #[cfg(unix)]
        {
            Ok(SeleneClient {
                handle_tx: unix_socket_handle::UnixSocketHandle::connect(event_callback)?,
            })
        }

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

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

trait SeleneIpcHandle {
    fn connect<F>(event_callback: F) -> Result<Sender<IpcRequest>, IpcHandleError>
    where
        Self: Sized,
        F: FnMut(PlayerEvent) + Send + Sync + 'static;

    fn reconnect(&mut self) -> Result<(), IpcHandleError>;

    fn run(self);
}

trait IpcReader: Read {
    fn try_read_packet_type(&mut self) -> io::Result<Option<PacketType>>;
}

struct SeleneIpcRunner {
    pending_ipc_callback: Option<CallbackFn>,
    event_callback: Box<dyn FnMut(PlayerEvent) + Send + Sync + 'static>,
}

impl SeleneIpcRunner {
    fn new<F>(event_callback: F) -> Self
    where
        F: FnMut(PlayerEvent) + Send + Sync + 'static,
    {
        Self {
            pending_ipc_callback: None,
            event_callback: Box::new(event_callback),
        }
    }

    fn accepting_ipc(&self) -> bool {
        self.pending_ipc_callback.is_none()
    }

    /// Runs a single cycle using the current handle
    ///
    /// Returns [`true`] if there was a disconnect
    fn run_cycle<R, W>(
        &mut self,
        command: Option<IpcRequest>,
        reader: &mut R,
        writer: &mut W,
    ) -> Result<(), ()>
    where
        R: IpcReader,
        W: Write,
    {
        if self.pending_ipc_callback.is_none()
            && let Some(command) = command
        {
            match command {
                IpcRequest::Ipc { command, callback } => {
                    if send_command(writer, &command).is_err() {
                        return Ok(());
                    }

                    if command.responds() {
                        self.pending_ipc_callback = callback;
                    }
                }
                IpcRequest::Reconnect { callback } => {
                    let _ = callback.send(Ok(()));
                }
            }
        }

        let packet_type = match reader.try_read_packet_type() {
            Ok(Some(packet_type)) => packet_type,
            Ok(None) => {
                wait();
                return Ok(());
            }
            Err(_) => return Err(()),
        };

        let Ok(data) = read_packet_data(reader) else {
            return Ok(());
        };

        match packet_type {
            PacketType::Unknown => panic!("Daemon sent unknown bytes"),
            PacketType::Event => {
                let event: PlayerEvent =
                    postcard::from_bytes(data.as_slice()).expect("Daemon sent corrupted bytes");

                (self.event_callback)(event);
            }
            PacketType::Response => {
                if let Some(ipc_callback) = self.pending_ipc_callback.take() {
                    ipc_callback(Ok(&data));
                }
            }
            PacketType::Error => {
                let err: PacketError =
                    postcard::from_bytes(data.as_slice()).expect("Daemon sent corrupted bytes");

                if let Some(ipc_callback) = self.pending_ipc_callback.take() {
                    ipc_callback(Err(err));
                }
            }
            PacketType::Disconnect => {
                if let Some(ipc_callback) = self.pending_ipc_callback.take() {
                    ipc_callback(Err(PacketError::Disconnect));
                }

                return Err(());
            }
        }

        Ok(())
    }
}

/// 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
fn read_packet_data(reader: &mut impl Read) -> io::Result<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    reader.read_exact(&mut len_buf)?;
    let len = u32::from_be_bytes(len_buf) as usize;

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

    Ok(data)
}

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