linux_cec/

async_support.rs

/*
 * Copyright © 2024 Valve Software
 * SPDX-License-Identifier: LGPL-2.1-or-later
 */

use linux_cec_sys::structs::{cec_caps, cec_msg};
use nix::poll::PollTimeout;
use std::ffi::OsString;
use std::fs::File;
use std::panic::resume_unwind;
use std::path::Path;
use std::sync::mpsc::{channel, Receiver, RecvError, SendError, Sender};
use std::thread::{self, JoinHandle};
use tokio::fs::OpenOptions;
use tokio::sync::oneshot;

use crate::device::{
    Capabilities, ConnectorInfo, Device, DevicePoller, Envelope, PollResult, PollStatus,
};
use crate::message::{Message, Opcode};
use crate::operand::UiCommand;
use crate::{
    Error, FollowerMode, InitiatorMode, LogicalAddress, LogicalAddressType, PhysicalAddress,
    Result, Timeout, VendorId,
};

macro_rules! relay {
    ($self:expr, $message:ident) => {{
        let (tx, rx) = oneshot::channel();
        $self.tx
            .send(DeviceCommand::$message(tx))?;
        rx.await?
    }};

    ($self:expr, $message:ident => $($args:expr),*) => {{
        let (tx, rx) = oneshot::channel();
        $self.tx
            .send(DeviceCommand::$message($($args,)* tx))?;
        rx.await?
    }};
}

type ResultChannel<T> = oneshot::Sender<Result<T>>;

enum DeviceCommand {
    Drop,
    GetPoller(ResultChannel<AsyncDevicePoller>),
    SetBlocking(bool, ResultChannel<()>),
    GetInitiatorMode(ResultChannel<InitiatorMode>),
    SetInitiatorMode(InitiatorMode, ResultChannel<()>),
    GetFollowerMode(ResultChannel<FollowerMode>),
    SetFollowerMode(FollowerMode, ResultChannel<()>),
    GetRawCapabilities(ResultChannel<cec_caps>),
    GetCapabilities(ResultChannel<Capabilities>),
    GetDriverName(ResultChannel<OsString>),
    GetAdapterName(ResultChannel<OsString>),
    GetPhysicalAddress(ResultChannel<PhysicalAddress>),
    SetPhysicalAddress(PhysicalAddress, ResultChannel<()>),
    GetLogicalAddresses(ResultChannel<Vec<LogicalAddress>>),
    SetLogicalAddresses(Vec<LogicalAddressType>, ResultChannel<()>),
    SetLogicalAddress(LogicalAddressType, ResultChannel<()>),
    ClearLogicalAddresses(ResultChannel<()>),
    GetOsdName(ResultChannel<OsString>),
    SetOsdName(String, ResultChannel<()>),
    GetVendorId(ResultChannel<Option<VendorId>>),
    SetVendorId(Option<VendorId>, ResultChannel<()>),
    TransmitMessage(Message, LogicalAddress, ResultChannel<u32>),
    TransmitRawMessage(cec_msg, ResultChannel<cec_msg>),
    TransmitReceiveMessage(
        Message,
        LogicalAddress,
        Opcode,
        Timeout,
        ResultChannel<Envelope>,
    ),
    ReceiveMessage(Timeout, ResultChannel<Envelope>),
    ReceiveRawMessage(u32, ResultChannel<cec_msg>),
    PollAddress(LogicalAddress, ResultChannel<()>),
    HandleStatus(PollStatus, ResultChannel<Vec<PollResult>>),
    GetConnectorInfo(ResultChannel<ConnectorInfo>),
    SetActiveSource(Option<PhysicalAddress>, ResultChannel<()>),
    Wake(bool, bool, ResultChannel<()>),
    Standby(LogicalAddress, ResultChannel<()>),
    PressUserControl(UiCommand, LogicalAddress, ResultChannel<()>),
    ReleaseUserControl(LogicalAddress, ResultChannel<()>),
}

/// An asynchronous version of [`Device`].
///
/// As this is simply an asynchronous interface for the standard [`Device`] struct,
/// refer to the documentation there for information about the individual methods.
#[derive(Debug)]
pub struct AsyncDevice {
    thread: Option<JoinHandle<Result<()>>>,
    tx: Sender<DeviceCommand>,
}

#[derive(Debug)]
pub struct AsyncDevicePoller {
    thread: Option<JoinHandle<Result<()>>>,
    tx: Sender<PollerCommand>,
}

enum PollerCommand {
    Drop,
    Poll(PollTimeout, ResultChannel<PollStatus>),
}

struct DeviceThread {
    device: Device,
    rx: Receiver<DeviceCommand>,
}

impl AsyncDevice {
    pub async fn open(path: impl AsRef<Path>) -> Result<AsyncDevice> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(false)
            .open(path)
            .await?;

        let file = file.into_std().await;
        AsyncDevice::try_from(file)
    }

    pub async fn set_blocking(&self, blocking: bool) -> Result<()> {
        relay! { self, SetBlocking => blocking }
    }

    pub async fn get_poller(&self) -> Result<AsyncDevicePoller> {
        relay! { self, GetPoller }
    }

    pub async fn poll(&self, timeout: PollTimeout) -> Result<Vec<PollResult>> {
        let poller = self.get_poller().await?;
        let status = poller.poll(timeout).await?;
        self.handle_status(status).await
    }

    pub async fn get_initiator_mode(&self) -> Result<InitiatorMode> {
        relay! { self, GetInitiatorMode }
    }

    pub async fn set_initiator_mode(&self, mode: InitiatorMode) -> Result<()> {
        relay! { self, SetInitiatorMode => mode }
    }

    pub async fn get_follower_mode(&self) -> Result<FollowerMode> {
        relay! { self, GetFollowerMode }
    }

    pub async fn set_follower_mode(&self, mode: FollowerMode) -> Result<()> {
        relay! { self, SetFollowerMode => mode }
    }

    pub async fn get_capabilities(&self) -> Result<Capabilities> {
        relay! { self, GetCapabilities }
    }

    pub async fn get_raw_capabilities(&self) -> Result<cec_caps> {
        relay! { self, GetRawCapabilities }
    }

    pub async fn get_driver_name(&self) -> Result<OsString> {
        relay! { self, GetDriverName }
    }

    pub async fn get_adapter_name(&self) -> Result<OsString> {
        relay! { self, GetAdapterName }
    }

    pub async fn get_physical_address(&self) -> Result<PhysicalAddress> {
        relay! { self, GetPhysicalAddress }
    }

    pub async fn set_physical_address(&self, phys_addr: PhysicalAddress) -> Result<()> {
        relay! { self, SetPhysicalAddress => phys_addr }
    }

    pub async fn get_logical_addresses(&self) -> Result<Vec<LogicalAddress>> {
        relay! { self, GetLogicalAddresses }
    }

    pub async fn set_logical_addresses(&self, log_addrs: &[LogicalAddressType]) -> Result<()> {
        relay! { self, SetLogicalAddresses => Vec::from(log_addrs) }
    }

    pub async fn set_logical_address(&self, log_addr: LogicalAddressType) -> Result<()> {
        relay! { self, SetLogicalAddress => log_addr }
    }

    pub async fn clear_logical_addresses(&self) -> Result<()> {
        relay! { self, ClearLogicalAddresses }
    }

    pub async fn get_osd_name(&self) -> Result<OsString> {
        relay! { self, GetOsdName }
    }

    pub async fn set_osd_name(&self, name: &str) -> Result<()> {
        relay! { self, SetOsdName => name.to_string() }
    }

    pub async fn get_vendor_id(&self) -> Result<Option<VendorId>> {
        relay! { self, GetVendorId }
    }

    pub async fn set_vendor_id(&self, vendor_id: Option<VendorId>) -> Result<()> {
        relay! { self, SetVendorId => vendor_id }
    }

    pub async fn tx_message(&self, message: &Message, destination: LogicalAddress) -> Result<u32> {
        relay! { self, TransmitMessage => *message, destination }
    }

    pub async fn tx_raw_message(&self, message: &mut cec_msg) -> Result<()> {
        let new_message = relay! { self, TransmitRawMessage => message.clone() }?;
        *message = new_message;
        Ok(())
    }

    pub async fn tx_rx_message(
        &self,
        message: &Message,
        destination: LogicalAddress,
        opcode: Opcode,
        timeout: Timeout,
    ) -> Result<Envelope> {
        relay! { self, TransmitReceiveMessage => *message, destination, opcode, timeout }
    }

    pub async fn rx_message(&self, timeout: Timeout) -> Result<Envelope> {
        relay! { self, ReceiveMessage => timeout }
    }

    pub async fn rx_raw_message(&self, timeout: u32) -> Result<cec_msg> {
        relay! { self, ReceiveRawMessage => timeout }
    }

    pub async fn poll_address(&self, destination: LogicalAddress) -> Result<()> {
        relay! { self, PollAddress => destination }
    }

    pub async fn handle_status(&self, status: PollStatus) -> Result<Vec<PollResult>> {
        relay! { self, HandleStatus => status }
    }

    pub async fn get_connector_info(&self) -> Result<ConnectorInfo> {
        relay! { self, GetConnectorInfo }
    }

    pub async fn set_active_source(&self, address: Option<PhysicalAddress>) -> Result<()> {
        relay! { self, SetActiveSource => address }
    }

    pub async fn wake(&self, set_active: bool, text_view: bool) -> Result<()> {
        relay! { self, Wake => set_active, text_view }
    }

    pub async fn standby(&self, target: LogicalAddress) -> Result<()> {
        relay! { self, Standby => target }
    }

    pub async fn press_user_control(
        &self,
        ui_command: UiCommand,
        target: LogicalAddress,
    ) -> Result<()> {
        relay! { self, PressUserControl => ui_command, target }
    }

    pub async fn release_user_control(&self, target: LogicalAddress) -> Result<()> {
        relay! { self, ReleaseUserControl => target }
    }

    /// Clean up the underlying [`Device`] struct. This is safer than just dropping it,
    /// as you can handle any errors that may occur.
    pub async fn close(mut self) -> Result<()> {
        self.tx.send(DeviceCommand::Drop)?;
        let Some(thread): Option<JoinHandle<Result<()>>> = self.thread.take() else {
            return Ok(());
        };
        match thread.join() {
            Ok(r) => r,
            Err(e) => resume_unwind(e),
        }
    }
}

impl From<Device> for AsyncDevice {
    fn from(device: Device) -> AsyncDevice {
        let (tx, rx) = channel();
        let mut thread = DeviceThread { device, rx };

        let thread = thread::spawn(move || thread.run());
        AsyncDevice {
            thread: Some(thread),
            tx,
        }
    }
}

impl TryFrom<File> for AsyncDevice {
    type Error = Error;

    fn try_from(file: File) -> Result<AsyncDevice> {
        let device = Device::try_from(file)?;
        Ok(AsyncDevice::from(device))
    }
}

impl Drop for AsyncDevice {
    fn drop(&mut self) {
        let Some(thread) = self.thread.take() else {
            return;
        };
        let _ = self.tx.send(DeviceCommand::Drop);
        let _ = thread.join();
    }
}

impl DeviceThread {
    fn run(&mut self) -> Result<()> {
        loop {
            match self.rx.recv()? {
                DeviceCommand::Drop => break,
                DeviceCommand::GetPoller(tx) => {
                    let _ = tx.send(self.device.get_poller().map(AsyncDevicePoller::from));
                }
                DeviceCommand::SetBlocking(block, tx) => {
                    let _ = tx.send(self.device.set_blocking(block));
                }
                DeviceCommand::GetInitiatorMode(tx) => {
                    let _ = tx.send(self.device.get_initiator_mode());
                }
                DeviceCommand::SetInitiatorMode(mode, tx) => {
                    let _ = tx.send(self.device.set_initiator_mode(mode));
                }
                DeviceCommand::GetFollowerMode(tx) => {
                    let _ = tx.send(self.device.get_follower_mode());
                }
                DeviceCommand::SetFollowerMode(mode, tx) => {
                    let _ = tx.send(self.device.set_follower_mode(mode));
                }
                DeviceCommand::GetCapabilities(tx) => {
                    let _ = tx.send(self.device.get_capabilities());
                }
                DeviceCommand::GetRawCapabilities(tx) => {
                    let _ = tx.send(self.device.get_raw_capabilities());
                }
                DeviceCommand::GetDriverName(tx) => {
                    let _ = tx.send(self.device.get_driver_name());
                }
                DeviceCommand::GetAdapterName(tx) => {
                    let _ = tx.send(self.device.get_adapter_name());
                }
                DeviceCommand::GetPhysicalAddress(tx) => {
                    let _ = tx.send(self.device.get_physical_address());
                }
                DeviceCommand::SetPhysicalAddress(phys_addr, tx) => {
                    let _ = tx.send(self.device.set_physical_address(phys_addr));
                }
                DeviceCommand::GetLogicalAddresses(tx) => {
                    let _ = tx.send(self.device.get_logical_addresses());
                }
                DeviceCommand::SetLogicalAddresses(log_addrs, tx) => {
                    let _ = tx.send(self.device.set_logical_addresses(&log_addrs));
                }
                DeviceCommand::SetLogicalAddress(log_addr, tx) => {
                    let _ = tx.send(self.device.set_logical_address(log_addr));
                }
                DeviceCommand::ClearLogicalAddresses(tx) => {
                    let _ = tx.send(self.device.clear_logical_addresses());
                }
                DeviceCommand::GetOsdName(tx) => {
                    let _ = tx.send(self.device.get_osd_name());
                }
                DeviceCommand::SetOsdName(name, tx) => {
                    let _ = tx.send(self.device.set_osd_name(&name));
                }
                DeviceCommand::GetVendorId(tx) => {
                    let _ = tx.send(self.device.get_vendor_id());
                }
                DeviceCommand::SetVendorId(vendor_id, tx) => {
                    let _ = tx.send(self.device.set_vendor_id(vendor_id));
                }
                DeviceCommand::TransmitMessage(message, dest, tx) => {
                    let _ = tx.send(self.device.tx_message(&message, dest));
                }
                DeviceCommand::TransmitRawMessage(msg, tx) => {
                    let mut msg = msg.clone();
                    let _ = tx.send(self.device.tx_raw_message(&mut msg).and(Ok(msg)));
                }
                DeviceCommand::TransmitReceiveMessage(message, dest, opcode, timeout, tx) => {
                    let _ = tx.send(self.device.tx_rx_message(&message, dest, opcode, timeout));
                }
                DeviceCommand::ReceiveMessage(timeout, tx) => {
                    let _ = tx.send(self.device.rx_message(timeout));
                }
                DeviceCommand::ReceiveRawMessage(timeout, tx) => {
                    let _ = tx.send(self.device.rx_raw_message(timeout));
                }
                DeviceCommand::PollAddress(dest, tx) => {
                    let _ = tx.send(self.device.poll_address(dest));
                }
                DeviceCommand::HandleStatus(status, tx) => {
                    let _ = tx.send(self.device.handle_status(status));
                }
                DeviceCommand::GetConnectorInfo(tx) => {
                    let _ = tx.send(self.device.get_connector_info());
                }
                DeviceCommand::SetActiveSource(address, tx) => {
                    let _ = tx.send(self.device.set_active_source(address));
                }
                DeviceCommand::Wake(set_active, text_view, tx) => {
                    let _ = tx.send(self.device.wake(set_active, text_view));
                }
                DeviceCommand::Standby(target, tx) => {
                    let _ = tx.send(self.device.standby(target));
                }
                DeviceCommand::PressUserControl(ui_command, target, tx) => {
                    let _ = tx.send(self.device.press_user_control(ui_command, target));
                }
                DeviceCommand::ReleaseUserControl(target, tx) => {
                    let _ = tx.send(self.device.release_user_control(target));
                }
            }
        }
        Ok(())
    }
}

impl From<RecvError> for Error {
    fn from(error: RecvError) -> Error {
        Error::UnknownError(error.to_string())
    }
}

impl<T> From<SendError<T>> for Error {
    fn from(error: SendError<T>) -> Error {
        Error::UnknownError(error.to_string())
    }
}

impl From<oneshot::error::RecvError> for Error {
    fn from(error: oneshot::error::RecvError) -> Error {
        Error::UnknownError(error.to_string())
    }
}

impl From<DevicePoller> for AsyncDevicePoller {
    fn from(poller: DevicePoller) -> AsyncDevicePoller {
        let (tx, rx) = channel();

        let thread = thread::spawn(move || {
            loop {
                match rx.recv()? {
                    PollerCommand::Drop => break,
                    PollerCommand::Poll(timeout, tx) => {
                        let _ = tx.send(poller.poll(timeout));
                    }
                }
            }
            Ok(())
        });
        AsyncDevicePoller {
            thread: Some(thread),
            tx,
        }
    }
}

impl AsyncDevicePoller {
    pub async fn poll(&self, timeout: PollTimeout) -> Result<PollStatus> {
        let (tx, rx) = oneshot::channel();
        self.tx.send(PollerCommand::Poll(timeout, tx))?;
        rx.await?
    }
}

impl Drop for AsyncDevicePoller {
    fn drop(&mut self) {
        let _ = self.tx.send(PollerCommand::Drop);
        let _ = self.thread.take().unwrap().join();
    }
}