mavlink-core 0.21.0

//! Async Serial MAVLink connection

use std::io;

use futures::lock::{Mutex, MutexGuard};
use tokio::io::{ReadHalf, WriteHalf};
use tokio_serial::{SerialPort, SerialPortBuilderExt, SerialStream};

use crate::AsyncMavlinkReader;
use crate::Dialect;
use crate::connection::r#async::{
    AsyncConnectable, AsyncDialectConnection, AsyncTransport, Transport,
};
use crate::connection::direct_serial::config::SerialConfig;

pub struct AsyncSerialConnection {
    read_port: Mutex<AsyncMavlinkReader<ReadHalf<SerialStream>>>,
    write_port: Mutex<WriteHalf<SerialStream>>,
}

impl AsyncTransport for AsyncSerialConnection {
    type Reader = ReadHalf<SerialStream>;
    type Writer = WriteHalf<SerialStream>;

    fn reader(&self) -> &Mutex<AsyncMavlinkReader<Self::Reader>> {
        &self.read_port
    }

    async fn lock_writer(&self) -> Option<MutexGuard<'_, Self::Writer>> {
        Some(self.write_port.lock().await)
    }

    fn retry_receive(&self, error: &crate::error::MessageReadError) -> bool {
        !matches!(error, crate::error::MessageReadError::Io(error) if error.kind() == io::ErrorKind::UnexpectedEof)
    }
}

impl SerialConfig {
    pub(crate) async fn open_async(&self) -> io::Result<AsyncSerialConnection> {
        let mut port = tokio_serial::new(&self.port_name, self.baud_rate).open_native_async()?;
        port.set_data_bits(tokio_serial::DataBits::Eight)?;
        port.set_parity(tokio_serial::Parity::None)?;
        port.set_stop_bits(tokio_serial::StopBits::One)?;
        port.set_flow_control(tokio_serial::FlowControl::None)?;

        let (reader, writer) = tokio::io::split(port);
        Ok(AsyncSerialConnection {
            read_port: Mutex::new(AsyncMavlinkReader::with_capacity(
                self.buffer_capacity(),
                reader,
            )),
            write_port: Mutex::new(writer),
        })
    }
}

impl AsyncConnectable for SerialConfig {
    async fn connect_async_with_dialect<D>(
        &self,
        dialect: D,
    ) -> io::Result<AsyncDialectConnection<D>>
    where
        D: Dialect + Send + Sync,
        D::Message: Send + Sync,
    {
        Ok(AsyncDialectConnection::new(
            Transport::Serial(self.open_async().await?),
            dialect,
        ))
    }
}