mavlink_core/connection/direct_serial/
async.rs

1//! Async Serial MAVLink connection
2
3use std::io;
4
5use futures::lock::{Mutex, MutexGuard};
6use tokio::io::{ReadHalf, WriteHalf};
7use tokio_serial::{SerialPort, SerialPortBuilderExt, SerialStream};
8
9use crate::AsyncMavlinkReader;
10use crate::Dialect;
11use crate::connection::r#async::{
12    AsyncConnectable, AsyncDialectConnection, AsyncTransport, Transport,
13};
14use crate::connection::direct_serial::config::SerialConfig;
15
16pub struct AsyncSerialConnection {
17    read_port: Mutex<AsyncMavlinkReader<ReadHalf<SerialStream>>>,
18    write_port: Mutex<WriteHalf<SerialStream>>,
19}
20
21impl AsyncTransport for AsyncSerialConnection {
22    type Reader = ReadHalf<SerialStream>;
23    type Writer = WriteHalf<SerialStream>;
24
25    fn reader(&self) -> &Mutex<AsyncMavlinkReader<Self::Reader>> {
26        &self.read_port
27    }
28
29    async fn lock_writer(&self) -> Option<MutexGuard<'_, Self::Writer>> {
30        Some(self.write_port.lock().await)
31    }
32
33    fn retry_receive(&self, error: &crate::error::MessageReadError) -> bool {
34        !matches!(error, crate::error::MessageReadError::Io(error) if error.kind() == io::ErrorKind::UnexpectedEof)
35    }
36}
37
38impl SerialConfig {
39    pub(crate) async fn open_async(&self) -> io::Result<AsyncSerialConnection> {
40        let mut port = tokio_serial::new(&self.port_name, self.baud_rate).open_native_async()?;
41        port.set_data_bits(tokio_serial::DataBits::Eight)?;
42        port.set_parity(tokio_serial::Parity::None)?;
43        port.set_stop_bits(tokio_serial::StopBits::One)?;
44        port.set_flow_control(tokio_serial::FlowControl::None)?;
45
46        let (reader, writer) = tokio::io::split(port);
47        Ok(AsyncSerialConnection {
48            read_port: Mutex::new(AsyncMavlinkReader::with_capacity(
49                self.buffer_capacity(),
50                reader,
51            )),
52            write_port: Mutex::new(writer),
53        })
54    }
55}
56
57impl AsyncConnectable for SerialConfig {
58    async fn connect_async_with_dialect<D>(
59        &self,
60        dialect: D,
61    ) -> io::Result<AsyncDialectConnection<D>>
62    where
63        D: Dialect + Send + Sync,
64        D::Message: Send + Sync,
65    {
66        Ok(AsyncDialectConnection::new(
67            Transport::Serial(self.open_async().await?),
68            dialect,
69        ))
70    }
71}
mavlink_core/connection/direct_serial/async.rs

mavlink_core/connection/direct_serial/
async.rs
