implement recv_raw for Connection wrappers into mavlink-core

mavlink-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mavlink-core"
-version = "0.13.2"
+version = "0.14.0"
 authors = [
     "Todd Stellanova",
     "Michal Podhradsky",
@@ -26,9 +26,15 @@ embedded-io-async = { version = "0.6.1", optional = true }
 serde = { version = "1.0.115", optional = true, features = ["derive"] }
 serde_arrays = { version = "0.1.0", optional = true }
 serial = { version = "0.4", optional = true }
-tokio = { version = "1.0", default-features = false, features = ["io-util", "net", "sync", "fs"], optional = true }
+tokio = { version = "1.0", default-features = false, features = [
+    "io-util",
+    "net",
+    "sync",
+    "fs",
+], optional = true }
 sha2 = { version = "0.10", optional = true }
 async-trait = { version = "0.1.18", optional = true }
+tokio-serial = { version = "5.4.4", default-features = false, optional = true }
 
 [features]
 "std" = ["byteorder/std"]
@@ -41,9 +47,16 @@ async-trait = { version = "0.1.18", optional = true }
 "embedded" = ["dep:embedded-io", "dep:embedded-io-async"]
 "embedded-hal-02" = ["dep:nb", "dep:embedded-hal-02"]
 "serde" = ["dep:serde", "dep:serde_arrays"]
-"tokio-1" = ["dep:tokio", "dep:async-trait"]
+"tokio-1" = ["dep:tokio", "dep:async-trait", "dep:tokio-serial"]
 "signing" = ["dep:sha2"]
-default = ["std", "tcp", "udp", "direct-serial", "serde"]
+default = ["std", "tcp", "udp", "direct-serial", "serde", "tokio-1"]
 
 [dev-dependencies]
-tokio = { version = "1.0", default-features = false, features = ["io-util", "net", "sync", "fs", "macros", "rt"] }
+tokio = { version = "1.0", default-features = false, features = [
+    "io-util",
+    "net",
+    "sync",
+    "fs",
+    "macros",
+    "rt",
+] }

mavlink-core/src/async_connection/direct_serial.rs

@@ -0,0 +1,152 @@
+//! Async Serial MAVLINK connection
+
+use core::ops::DerefMut;
+use std::io;
+
+use tokio::sync::Mutex;
+use tokio_serial::{SerialPort, SerialPortBuilderExt, SerialStream};
+
+use crate::{
+    async_peek_reader::AsyncPeekReader, read_v1_raw_message_async, read_v2_raw_message_async,
+    read_v2_raw_message_async_signed, MAVLinkRawMessage, MAVLinkV2MessageRaw, MavHeader,
+    MavlinkVersion, Message,
+};
+
+#[cfg(not(feature = "signing"))]
+use crate::{read_versioned_msg_async, write_versioned_msg_async};
+#[cfg(feature = "signing")]
+use crate::{
+    read_versioned_msg_async_signed, write_versioned_msg_async_signed, SigningConfig, SigningData,
+};
+
+use super::AsyncMavConnection;
+
+pub fn open(settings: &str) -> io::Result<AsyncSerialConnection> {
+    let settings_toks: Vec<&str> = settings.split(':').collect();
+    if settings_toks.len() < 2 {
+        return Err(io::Error::new(
+            io::ErrorKind::AddrNotAvailable,
+            "Incomplete port settings",
+        ));
+    }
+
+    let Ok(baud) = settings_toks[1].parse::<u32>() else {
+        return Err(io::Error::new(
+            io::ErrorKind::AddrNotAvailable,
+            "Invalid baud rate",
+        ));
+    };
+
+    let port_name = settings_toks[0];
+    let mut port = tokio_serial::new(port_name, baud).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)?;
+
+    Ok(AsyncSerialConnection {
+        port: Mutex::new(AsyncPeekReader::new(port)),
+        sequence: Mutex::new(0),
+        protocol_version: MavlinkVersion::V2,
+        #[cfg(feature = "signing")]
+        signing_data: None,
+    })
+}
+
+pub struct AsyncSerialConnection {
+    port: Mutex<AsyncPeekReader<SerialStream>>,
+    sequence: Mutex<u8>,
+    protocol_version: MavlinkVersion,
+    #[cfg(feature = "signing")]
+    signing_data: Option<SigningData>,
+}
+
+#[async_trait::async_trait]
+impl<M: Message + Sync + Send> AsyncMavConnection<M> for AsyncSerialConnection {
+    async fn recv(&self) -> Result<(MavHeader, M), crate::error::MessageReadError> {
+        let mut port = self.port.lock().await;
+
+        #[cfg(not(feature = "signing"))]
+        let result = read_versioned_msg_async(port.deref_mut(), self.protocol_version).await;
+        #[cfg(feature = "signing")]
+        let result = read_versioned_msg_async_signed(
+            port.deref_mut(),
+            self.protocol_version,
+            self.signing_data.as_ref(),
+        )
+        .await;
+        result
+    }
+
+    async fn recv_raw(&self) -> Result<MAVLinkRawMessage, crate::error::MessageReadError> {
+        let mut port = self.port.lock().await;
+        #[cfg(not(feature = "signing"))]
+        let result = match self.protocol_version {
+            MavlinkVersion::V1 => {
+                MAVLinkRawMessage::V1(read_v1_raw_message_async::<M, _>(port.deref_mut()).await?)
+            }
+            MavlinkVersion::V2 => {
+                MAVLinkRawMessage::V2(read_v2_raw_message_async::<M, _>(port.deref_mut()).await?)
+            }
+        };
+        #[cfg(feature = "signing")]
+        let result = match self.protocol_version {
+            MavlinkVersion::V1 => {
+                MAVLinkRawMessage::V1(read_v1_raw_message_async::<M, _>(port.deref_mut()).await?)
+            }
+            MavlinkVersion::V2 => MAVLinkRawMessage::V2(
+                read_v2_raw_message_async_signed::<M, _>(
+                    port.deref_mut(),
+                    self.signing_data.as_ref(),
+                )
+                .await?,
+            ),
+        };
+
+        Ok(result)
+    }
+
+    async fn send(
+        &self,
+        header: &MavHeader,
+        data: &M,
+    ) -> Result<usize, crate::error::MessageWriteError> {
+        let mut port = self.port.lock().await;
+        let mut sequence = self.sequence.lock().await;
+
+        let header = MavHeader {
+            sequence: *sequence,
+            system_id: header.system_id,
+            component_id: header.component_id,
+        };
+
+        *sequence = sequence.wrapping_add(1);
+
+        #[cfg(not(feature = "signing"))]
+        let result =
+            write_versioned_msg_async(port.reader_mut(), self.protocol_version, header, data).await;
+        #[cfg(feature = "signing")]
+        let result = write_versioned_msg_async_signed(
+            port.reader_mut(),
+            self.protocol_version,
+            header,
+            data,
+            self.signing_data.as_ref(),
+        )
+        .await;
+        result
+    }
+
+    fn set_protocol_version(&mut self, version: MavlinkVersion) {
+        self.protocol_version = version;
+    }
+
+    fn get_protocol_version(&self) -> MavlinkVersion {
+        self.protocol_version
+    }
+
+    #[cfg(feature = "signing")]
+    fn setup_signing(&mut self, signing_data: Option<SigningConfig>) {
+        self.signing_data = signing_data.map(SigningData::from_config)
+    }
+}

mavlink-core/src/async_connection/file.rs

@@ -6,6 +6,10 @@
 use crate::error::{MessageReadError, MessageWriteError};
 
 use crate::{async_peek_reader::AsyncPeekReader, MavHeader, MavlinkVersion, Message};
+use crate::{
+    read_v1_raw_message_async, read_v2_raw_message_async, read_v2_raw_message_async_signed,
+    MAVLinkRawMessage, MAVLinkV2MessageRaw,
+};
 
 use tokio::fs::File;
 use tokio::io;
@@ -64,6 +68,34 @@
         }
     }
 
+    async fn recv_raw(&self) -> Result<MAVLinkRawMessage, crate::error::MessageReadError> {
+        let mut file = self.file.lock().await;
+        #[cfg(not(feature = "signing"))]
+        let result = match self.protocol_version {
+            MavlinkVersion::V1 => {
+                MAVLinkRawMessage::V1(read_v1_raw_message_async::<M, _>(file.deref_mut()).await?)
+            }
+            MavlinkVersion::V2 => {
+                MAVLinkRawMessage::V2(read_v2_raw_message_async::<M, _>(file.deref_mut()).await?)
+            }
+        };
+        #[cfg(feature = "signing")]
+        let result = match self.protocol_version {
+            MavlinkVersion::V1 => {
+                MAVLinkRawMessage::V1(read_v1_raw_message_async::<M, _>(file.deref_mut()).await?)
+            }
+            MavlinkVersion::V2 => MAVLinkRawMessage::V2(
+                read_v2_raw_message_async_signed::<M, _>(
+                    file.deref_mut(),
+                    self.signing_data.as_ref(),
+                )
+                .await?,
+            ),
+        };
+
+        Ok(result)
+    }
+
     async fn send(&self, _header: &MavHeader, _data: &M) -> Result<usize, MessageWriteError> {
         Ok(0)
     }

mavlink-core/src/async_connection/mod.rs

@@ -1,13 +1,16 @@
 use tokio::io;
 
-use crate::{MavFrame, MavHeader, MavlinkVersion, Message};
+use crate::{MAVLinkRawMessage, MAVLinkV2MessageRaw, MavFrame, MavHeader, MavlinkVersion, Message};
 
 #[cfg(feature = "tcp")]
 mod tcp;
 
 #[cfg(feature = "udp")]
 mod udp;
 
+#[cfg(feature = "direct-serial")]
+mod direct_serial;
+
 mod file;
 
 #[cfg(feature = "signing")]
@@ -21,6 +24,8 @@
     /// Yield until a valid frame is received, ignoring invalid messages.
     async fn recv(&self) -> Result<(MavHeader, M), crate::error::MessageReadError>;
 
+    async fn recv_raw(&self) -> Result<MAVLinkRawMessage, crate::error::MessageReadError>;
+
     /// Send a mavlink message
     async fn send(
         &self,
@@ -70,9 +75,9 @@
 ///  * `udpin:<addr>:<port>` to create a UDP server, listening for incoming packets
 ///  * `udpout:<addr>:<port>` to create a UDP client
 ///  * `udpbcast:<addr>:<port>` to create a UDP broadcast
+///  * `serial:<port>:<baudrate>` to create a serial connection
 ///  * `file:<path>` to extract file data
 ///
-/// Serial is currently not supported for async connections, use [`crate::connect`] instead.
 /// The type of the connection is determined at runtime based on the address type, so the
 /// connection is returned as a trait object.
 pub async fn connect_async<M: Message + Sync + Send>(
@@ -101,6 +106,15 @@
         {
             protocol_err
         }
+    } else if cfg!(feature = "direct-serial") && address.starts_with("serial") {
+        #[cfg(feature = "direct-serial")]
+        {
+            Ok(Box::new(direct_serial::open(&address["serial:".len()..])?))
+        }
+        #[cfg(not(feature = "direct-serial"))]
+        {
+            protocol_err
+        }
     } else if address.starts_with("file") {
         Ok(Box::new(file::open(&address["file:".len()..]).await?))
     } else {

mavlink-core/src/async_connection/tcp.rs

@@ -2,7 +2,11 @@
 
 use super::{get_socket_addr, AsyncMavConnection};
 use crate::async_peek_reader::AsyncPeekReader;
-use crate::{MavHeader, MavlinkVersion, Message};
+use crate::{
+    read_v1_raw_message, read_v1_raw_message_async, read_v2_raw_message_async,
+    read_v2_raw_message_async_signed, MAVLinkRawMessage, MAVLinkV2MessageRaw, MavHeader,
+    MavlinkVersion, Message,
+};
 
 use core::ops::DerefMut;
 use tokio::io;
@@ -112,6 +116,34 @@
         result
     }
 
+    async fn recv_raw(&self) -> Result<MAVLinkRawMessage, crate::error::MessageReadError> {
+        let mut reader = self.reader.lock().await;
+        #[cfg(not(feature = "signing"))]
+        let result = match self.protocol_version {
+            MavlinkVersion::V1 => {
+                MAVLinkRawMessage::V1(read_v1_raw_message_async::<M, _>(reader.deref_mut()).await?)
+            }
+            MavlinkVersion::V2 => {
+                MAVLinkRawMessage::V2(read_v2_raw_message_async::<M, _>(reader.deref_mut()).await?)
+            }
+        };
+        #[cfg(feature = "signing")]
+        let result = match self.protocol_version {
+            MavlinkVersion::V1 => {
+                MAVLinkRawMessage::V1(read_v1_raw_message_async::<M, _>(reader.deref_mut()).await?)
+            }
+            MavlinkVersion::V2 => MAVLinkRawMessage::V2(
+                read_v2_raw_message_async_signed::<M, _>(
+                    reader.deref_mut(),
+                    self.signing_data.as_ref(),
+                )
+                .await?,
+            ),
+        };
+
+        Ok(result)
+    }
+
     async fn send(
         &self,
         header: &MavHeader,