Add tokio-based async read/write

mavlink-core/Cargo.toml

@@ -26,6 +26,7 @@ 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"], optional = true }
 
 [features]
 "std" = ["byteorder/std"]
@@ -38,4 +39,5 @@ serial = { version = "0.4", 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"]
 default = ["std", "tcp", "udp", "direct-serial", "serde"]

mavlink-core/src/lib.rs

@@ -511,7 +511,7 @@ impl MAVLinkV2MessageRaw {
     }
 
     #[inline]
-    pub fn header(&mut self) -> &[u8] {
+    pub fn header(&self) -> &[u8] {
         &self.0[1..=Self::HEADER_SIZE]
     }
 
@@ -698,6 +698,41 @@ pub fn read_v2_raw_message<M: Message, R: Read>(
     }
 }
 
+/// Async read a raw buffer with the mavlink message
+/// V2 maximum size is 280 bytes: `<https://mavlink.io/en/guide/serialization.html>`
+#[cfg(feature = "tokio-1")]
+pub async fn read_v2_raw_message_async<M: Message, R: tokio::io::AsyncReadExt + Unpin>(
+    reader: &mut R,
+) -> Result<MAVLinkV2MessageRaw, error::MessageReadError> {
+    loop {
+        loop {
+            // search for the magic framing value indicating start of mavlink message
+            if reader.read_u8().await? == MAV_STX_V2 {
+                break;
+            }
+        }
+
+        let mut message = MAVLinkV2MessageRaw::new();
+
+        message.0[0] = MAV_STX_V2;
+        let header_len = reader.read_exact(message.mut_header()).await?;
+        assert_eq!(header_len, MAVLinkV2MessageRaw::HEADER_SIZE);
+
+        if message.incompatibility_flags() & !MAVLINK_SUPPORTED_IFLAGS > 0 {
+            // if there are incompatibility flags set that we do not know discard the message
+            continue;
+        }
+
+        reader
+            .read_exact(message.mut_payload_and_checksum_and_sign())
+            .await?;
+
+        if message.has_valid_crc::<M>() {
+            return Ok(message);
+        }
+    }
+}
+
 /// Async read a raw buffer with the mavlink message
 /// V2 maximum size is 280 bytes: `<https://mavlink.io/en/guide/serialization.html>`
 ///
@@ -766,6 +801,27 @@ pub fn read_v2_msg<M: Message, R: Read>(
         .map_err(|err| err.into())
 }
 
+/// Async read a MAVLink v2  message from a Read stream.
+#[cfg(feature = "tokio-1")]
+pub async fn read_v2_msg_async<M: Message, R: tokio::io::AsyncReadExt + Unpin>(
+    read: &mut R,
+) -> Result<(MavHeader, M), error::MessageReadError> {
+    let message = read_v2_raw_message_async::<M, _>(read).await?;
+
+    M::parse(MavlinkVersion::V2, message.message_id(), message.payload())
+        .map(|msg| {
+            (
+                MavHeader {
+                    sequence: message.sequence(),
+                    system_id: message.system_id(),
+                    component_id: message.component_id(),
+                },
+                msg,
+            )
+        })
+        .map_err(|err| err.into())
+}
+
 /// Async read a MAVLink v2  message from a Read stream.
 ///
 /// NOTE: it will be add ~80KB to firmware flash size because all *_DATA::deser methods will be add to firmware.
@@ -807,6 +863,20 @@ pub fn write_versioned_msg<M: Message, W: Write>(
     }
 }
 
+/// Async write a message using the given mavlink version
+#[cfg(feature = "tokio-1")]
+pub async fn write_versioned_msg_async<M: Message, W: tokio::io::AsyncWriteExt + Unpin>(
+    w: &mut W,
+    version: MavlinkVersion,
+    header: MavHeader,
+    data: &M,
+) -> Result<usize, error::MessageWriteError> {
+    match version {
+        MavlinkVersion::V2 => write_v2_msg_async(w, header, data).await,
+        MavlinkVersion::V1 => write_v1_msg_async(w, header, data).await,
+    }
+}
+
 /// Async write a message using the given mavlink version
 ///
 /// NOTE: it will be add ~70KB to firmware flash size because all *_DATA::ser methods will be add to firmware.
@@ -841,6 +911,24 @@ pub fn write_v2_msg<M: Message, W: Write>(
     Ok(len)
 }
 
+/// Async write a MAVLink v2 message to a Write stream.
+#[cfg(feature = "tokio-1")]
+pub async fn write_v2_msg_async<M: Message, W: tokio::io::AsyncWriteExt + Unpin>(
+    w: &mut W,
+    header: MavHeader,
+    data: &M,
+) -> Result<usize, error::MessageWriteError> {
+    let mut message_raw = MAVLinkV2MessageRaw::new();
+    message_raw.serialize_message(header, data);
+
+    let payload_length: usize = message_raw.payload_length().into();
+    let len = 1 + MAVLinkV2MessageRaw::HEADER_SIZE + payload_length + 2;
+
+    w.write_all(&message_raw.0[..len]).await?;
+
+    Ok(len)
+}
+
 /// Async write a MAVLink v2 message to a Write stream.
 ///
 /// NOTE: it will be add ~70KB to firmware flash size because all *_DATA::ser methods will be add to firmware.
@@ -881,6 +969,24 @@ pub fn write_v1_msg<M: Message, W: Write>(
     Ok(len)
 }
 
+/// Async write a MAVLink v1 message to a Write stream.
+#[cfg(feature = "tokio-1")]
+pub async fn write_v1_msg_async<M: Message, W: tokio::io::AsyncWriteExt + Unpin>(
+    w: &mut W,
+    header: MavHeader,
+    data: &M,
+) -> Result<usize, error::MessageWriteError> {
+    let mut message_raw = MAVLinkV1MessageRaw::new();
+    message_raw.serialize_message(header, data);
+
+    let payload_length: usize = message_raw.payload_length().into();
+    let len = 1 + MAVLinkV1MessageRaw::HEADER_SIZE + payload_length + 2;
+
+    w.write_all(&message_raw.0[..len]).await?;
+
+    Ok(len)
+}
+
 /// Write a MAVLink v1 message to a Write stream.
 ///
 /// NOTE: it will be add ~70KB to firmware flash size because all *_DATA::ser methods will be add to firmware.

mavlink/Cargo.toml

@@ -19,15 +19,15 @@ edition = "2018"
 rust-version = "1.65.0"
 
 [build-dependencies]
-mavlink-bindgen = { version = "0.13.1", default-features = false }
+mavlink-bindgen = { path = "../mavlink-bindgen", default-features = false }
 
 [[example]]
 name = "mavlink-dump"
 path = "examples/mavlink-dump/src/main.rs"
 required-features = ["ardupilotmega"]
 
 [dependencies]
-mavlink-core = { version = "0.13.1", default-features = false }
+mavlink-core = { path = "../mavlink-core", default-features = false }
 num-traits = { workspace = true, default-features = false }
 num-derive = { workspace = true }
 bitflags = { workspace = true }
@@ -99,6 +99,7 @@ serde_arrays = { version = "0.1.0", optional = true }
 "embedded" = ["mavlink-core/embedded"]
 "embedded-hal-02" = ["mavlink-core/embedded-hal-02"]
 "serde" = ["mavlink-core/serde", "dep:serde", "dep:serde_arrays"]
+"tokio-1" = ["mavlink-core/tokio-1"]
 default = ["std", "tcp", "udp", "direct-serial", "serde", "ardupilotmega"]
 
 # build with all features on docs.rs so that users viewing documentation