mavlink_core/
async_reader.rs

1//! Incremental asynchronous MAVLink reader.
2
3use crate::{
4    MAVLinkMessageRaw, MavHeader, MavlinkVersion, Message,
5    error::MessageReadError,
6    frame_decoder::{FrameDecoder, VersionFilter},
7    reader::{try_decode_message, try_decode_raw_message},
8    static_dialect::StaticDialect,
9};
10
11#[cfg(feature = "tokio")]
12use crate::{Dialect, SigningData};
13
14#[cfg(all(feature = "embedded", not(feature = "std")))]
15use embedded_io_async::Read;
16#[cfg(feature = "tokio")]
17use tokio::io::{AsyncRead, AsyncReadExt};
18
19/// Incrementally reads complete MAVLink frames from an asynchronous byte stream.
20///
21/// Partial input remains buffered if a read is cancelled. Use the same MAVLink
22/// dialect `M` for the lifetime of a stream, because CRC validation is
23/// dialect-specific.
24pub struct AsyncMavlinkReader<R> {
25    source: R,
26    decoder: FrameDecoder,
27}
28
29impl<R> AsyncMavlinkReader<R> {
30    /// Creates a reader around `source` with the default read-ahead capacity.
31    ///
32    /// The buffer is allocated once and reused for the reader's lifetime.
33    #[cfg(feature = "std")]
34    pub fn new(source: R) -> Self
35    where
36        R: AsyncRead + Unpin,
37    {
38        Self {
39            source,
40            decoder: FrameDecoder::new(),
41        }
42    }
43
44    /// Creates an allocation-free reader around `source`.
45    #[cfg(not(feature = "std"))]
46    pub const fn new(source: R) -> Self
47    where
48        R: Read,
49    {
50        Self {
51            source,
52            decoder: FrameDecoder::new(),
53        }
54    }
55
56    /// Creates a reader with at least `capacity` bytes of read-ahead space.
57    ///
58    /// Capacities smaller than one maximum-size MAVLink frame are raised to
59    /// [`consts::MAX_FRAME_SIZE`](crate::consts::MAX_FRAME_SIZE). The buffer is
60    /// allocated once and reused for the lifetime of the reader.
61    #[cfg(feature = "std")]
62    pub fn with_capacity(capacity: usize, source: R) -> Self
63    where
64        R: AsyncRead + Unpin,
65    {
66        Self {
67            source,
68            decoder: FrameDecoder::with_capacity(capacity),
69        }
70    }
71
72    /// Returns a shared reference to the underlying source.
73    pub const fn get_ref(&self) -> &R {
74        &self.source
75    }
76
77    /// Returns a mutable reference to the underlying source.
78    ///
79    /// Reading directly from the source can skip bytes already buffered by
80    /// this reader and should therefore be avoided.
81    pub const fn get_mut(&mut self) -> &mut R {
82        &mut self.source
83    }
84
85    /// Returns the underlying source.
86    ///
87    /// Any bytes read ahead by this reader are discarded.
88    pub fn into_inner(self) -> R {
89        self.source
90    }
91}
92
93#[cfg(feature = "tokio")]
94impl<R: AsyncRead + Unpin> AsyncMavlinkReader<R> {
95    /// Reads and parses the next CRC-valid message accepted by `version`.
96    pub async fn read_message<M: Message>(
97        &mut self,
98        version: MavlinkVersion,
99    ) -> Result<(MavHeader, M), MessageReadError> {
100        self.read_message_inner(VersionFilter::Exact(version), None)
101            .await
102    }
103
104    /// Like [`Self::read_message`], using `dialect` for CRC validation and decoding.
105    pub async fn read_message_with_dialect<D: Dialect>(
106        &mut self,
107        dialect: &D,
108        version: MavlinkVersion,
109    ) -> Result<(MavHeader, D::Message), MessageReadError> {
110        self.read_message_with_dialect_inner(VersionFilter::Exact(version), dialect, None)
111            .await
112    }
113
114    /// Reads and parses the next CRC-valid MAVLink 1 or MAVLink 2 message.
115    pub async fn read_any_message<M: Message>(
116        &mut self,
117    ) -> Result<(MavHeader, M), MessageReadError> {
118        self.read_message_inner(VersionFilter::Any, None).await
119    }
120
121    /// Like [`Self::read_any_message`], using `dialect` for CRC validation and decoding.
122    pub async fn read_any_message_with_dialect<D: Dialect>(
123        &mut self,
124        dialect: &D,
125    ) -> Result<(MavHeader, D::Message), MessageReadError> {
126        self.read_message_with_dialect_inner(VersionFilter::Any, dialect, None)
127            .await
128    }
129
130    /// Reads the next CRC-valid raw message accepted by `version`.
131    pub async fn read_raw_message<M: Message>(
132        &mut self,
133        version: MavlinkVersion,
134    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
135        self.read_raw_message_inner::<M>(VersionFilter::Exact(version), None)
136            .await
137    }
138
139    /// Like [`Self::read_raw_message`], using `dialect` for CRC validation and decoding.
140    pub async fn read_raw_message_with_dialect<D: Dialect>(
141        &mut self,
142        dialect: &D,
143        version: MavlinkVersion,
144    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
145        self.read_raw_message_with_dialect_inner(VersionFilter::Exact(version), dialect, None)
146            .await
147    }
148
149    /// Reads the next CRC-valid MAVLink 1 or MAVLink 2 raw message.
150    pub async fn read_any_raw_message<M: Message>(
151        &mut self,
152    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
153        self.read_raw_message_inner::<M>(VersionFilter::Any, None)
154            .await
155    }
156
157    /// Like [`Self::read_any_raw_message`], using `dialect` for CRC validation and decoding.
158    pub async fn read_any_raw_message_with_dialect<D: Dialect>(
159        &mut self,
160        dialect: &D,
161    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
162        self.read_raw_message_with_dialect_inner(VersionFilter::Any, dialect, None)
163            .await
164    }
165
166    /// Reads, verifies, and parses the next message accepted by `version`.
167    /// With signing data, MAVLink 1 and unsigned MAVLink 2 frames require
168    /// `SigningConfig::allow_unsigned`; signed MAVLink 2 frames must have a valid
169    /// signature. With `None`, signature verification is skipped.
170    #[cfg(feature = "mav2-message-signing")]
171    pub async fn read_message_signed<M: Message>(
172        &mut self,
173        version: MavlinkVersion,
174        signing_data: Option<&SigningData>,
175    ) -> Result<(MavHeader, M), MessageReadError> {
176        self.read_message_inner(VersionFilter::Exact(version), signing_data)
177            .await
178    }
179
180    /// Like [`Self::read_message_signed`], using `dialect` for CRC validation and decoding.
181    #[cfg(feature = "mav2-message-signing")]
182    pub async fn read_message_signed_with_dialect<D: Dialect>(
183        &mut self,
184        dialect: &D,
185        version: MavlinkVersion,
186        signing_data: Option<&SigningData>,
187    ) -> Result<(MavHeader, D::Message), MessageReadError> {
188        self.read_message_with_dialect_inner(VersionFilter::Exact(version), dialect, signing_data)
189            .await
190    }
191
192    /// Reads, verifies, and parses the next MAVLink 1 or MAVLink 2 message.
193    /// With signing data, MAVLink 1 frames follow `SigningConfig::allow_unsigned`.
194    /// With `None`, signature verification is skipped.
195    #[cfg(feature = "mav2-message-signing")]
196    pub async fn read_any_message_signed<M: Message>(
197        &mut self,
198        signing_data: Option<&SigningData>,
199    ) -> Result<(MavHeader, M), MessageReadError> {
200        self.read_message_inner(VersionFilter::Any, signing_data)
201            .await
202    }
203
204    /// Like [`Self::read_any_message_signed`], using `dialect` for CRC validation and decoding.
205    #[cfg(feature = "mav2-message-signing")]
206    pub async fn read_any_message_signed_with_dialect<D: Dialect>(
207        &mut self,
208        dialect: &D,
209        signing_data: Option<&SigningData>,
210    ) -> Result<(MavHeader, D::Message), MessageReadError> {
211        self.read_message_with_dialect_inner(VersionFilter::Any, dialect, signing_data)
212            .await
213    }
214
215    /// Reads and verifies the next raw message accepted by `version`.
216    /// With signing data, MAVLink 1 and unsigned MAVLink 2 frames require
217    /// `SigningConfig::allow_unsigned`; signed MAVLink 2 frames must have a valid
218    /// signature. With `None`, signature verification is skipped.
219    #[cfg(feature = "mav2-message-signing")]
220    pub async fn read_raw_message_signed<M: Message>(
221        &mut self,
222        version: MavlinkVersion,
223        signing_data: Option<&SigningData>,
224    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
225        self.read_raw_message_inner::<M>(VersionFilter::Exact(version), signing_data)
226            .await
227    }
228
229    /// Like [`Self::read_raw_message_signed`], using `dialect` for CRC validation and decoding.
230    #[cfg(feature = "mav2-message-signing")]
231    pub async fn read_raw_message_signed_with_dialect<D: Dialect>(
232        &mut self,
233        dialect: &D,
234        version: MavlinkVersion,
235        signing_data: Option<&SigningData>,
236    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
237        self.read_raw_message_with_dialect_inner(
238            VersionFilter::Exact(version),
239            dialect,
240            signing_data,
241        )
242        .await
243    }
244
245    /// Reads and verifies the next MAVLink 1 or MAVLink 2 raw message.
246    /// With signing data, MAVLink 1 frames follow `SigningConfig::allow_unsigned`.
247    /// With `None`, signature verification is skipped.
248    #[cfg(feature = "mav2-message-signing")]
249    pub async fn read_any_raw_message_signed<M: Message>(
250        &mut self,
251        signing_data: Option<&SigningData>,
252    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
253        self.read_raw_message_inner::<M>(VersionFilter::Any, signing_data)
254            .await
255    }
256
257    /// Like [`Self::read_any_raw_message_signed`], using `dialect` for CRC validation and decoding.
258    #[cfg(feature = "mav2-message-signing")]
259    pub async fn read_any_raw_message_signed_with_dialect<D: Dialect>(
260        &mut self,
261        dialect: &D,
262        signing_data: Option<&SigningData>,
263    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
264        self.read_raw_message_with_dialect_inner(VersionFilter::Any, dialect, signing_data)
265            .await
266    }
267
268    #[inline]
269    pub(crate) async fn read_message_inner<M: Message>(
270        &mut self,
271        filter: VersionFilter,
272        signing_data: Option<&SigningData>,
273    ) -> Result<(MavHeader, M), MessageReadError> {
274        self.read_message_with_dialect_inner(filter, &StaticDialect::<M>::new(), signing_data)
275            .await
276    }
277
278    #[inline]
279    pub(crate) async fn read_message_with_dialect_inner<D: Dialect>(
280        &mut self,
281        filter: VersionFilter,
282        dialect: &D,
283        signing_data: Option<&SigningData>,
284    ) -> Result<(MavHeader, D::Message), MessageReadError> {
285        loop {
286            if let Some(result) =
287                try_decode_message(&mut self.decoder, filter, dialect, signing_data)
288            {
289                return result;
290            }
291            self.read_more().await?;
292        }
293    }
294
295    #[inline]
296    pub(crate) async fn read_raw_message_inner<M: Message>(
297        &mut self,
298        filter: VersionFilter,
299        signing_data: Option<&SigningData>,
300    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
301        self.read_raw_message_with_dialect_inner(filter, &StaticDialect::<M>::new(), signing_data)
302            .await
303    }
304
305    #[inline]
306    pub(crate) async fn read_raw_message_with_dialect_inner<D: Dialect>(
307        &mut self,
308        filter: VersionFilter,
309        dialect: &D,
310        signing_data: Option<&SigningData>,
311    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
312        loop {
313            if let Some(message) =
314                try_decode_raw_message(&mut self.decoder, filter, dialect, signing_data)
315            {
316                return Ok(message);
317            }
318            self.read_more().await?;
319        }
320    }
321
322    async fn read_more(&mut self) -> Result<(), MessageReadError> {
323        let destination = self.decoder.spare_capacity_mut();
324        assert!(
325            !destination.is_empty(),
326            "a pending MAVLink frame must fit in the decoder buffer"
327        );
328        let count = loop {
329            match self.source.read(destination).await {
330                Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
331                result => break result?,
332            }
333        };
334        if count == 0 {
335            return Err(MessageReadError::eof());
336        }
337        self.decoder.commit(count);
338        Ok(())
339    }
340}
341
342#[cfg(all(feature = "embedded", not(feature = "std")))]
343impl<R: Read> AsyncMavlinkReader<R> {
344    /// Reads and parses the next CRC-valid message accepted by `version`.
345    pub async fn read_message<M: Message>(
346        &mut self,
347        version: MavlinkVersion,
348    ) -> Result<(MavHeader, M), MessageReadError> {
349        let filter = VersionFilter::Exact(version);
350        loop {
351            if let Some(result) =
352                try_decode_message(&mut self.decoder, filter, &StaticDialect::<M>::new(), None)
353            {
354                return result;
355            }
356            self.read_more().await?;
357        }
358    }
359
360    /// Reads and parses the next CRC-valid MAVLink 1 or MAVLink 2 message.
361    pub async fn read_any_message<M: Message>(
362        &mut self,
363    ) -> Result<(MavHeader, M), MessageReadError> {
364        loop {
365            if let Some(result) = try_decode_message(
366                &mut self.decoder,
367                VersionFilter::Any,
368                &StaticDialect::<M>::new(),
369                None,
370            ) {
371                return result;
372            }
373            self.read_more().await?;
374        }
375    }
376
377    /// Reads the next CRC-valid raw message accepted by `version`.
378    pub async fn read_raw_message<M: Message>(
379        &mut self,
380        version: MavlinkVersion,
381    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
382        let filter = VersionFilter::Exact(version);
383        loop {
384            if let Some(message) =
385                try_decode_raw_message(&mut self.decoder, filter, &StaticDialect::<M>::new(), None)
386            {
387                return Ok(message);
388            }
389            self.read_more().await?;
390        }
391    }
392
393    /// Reads the next CRC-valid MAVLink 1 or MAVLink 2 raw message.
394    pub async fn read_any_raw_message<M: Message>(
395        &mut self,
396    ) -> Result<MAVLinkMessageRaw, MessageReadError> {
397        loop {
398            if let Some(message) = try_decode_raw_message(
399                &mut self.decoder,
400                VersionFilter::Any,
401                &StaticDialect::<M>::new(),
402                None,
403            ) {
404                return Ok(message);
405            }
406            self.read_more().await?;
407        }
408    }
409
410    async fn read_more(&mut self) -> Result<(), MessageReadError> {
411        let destination = self.decoder.spare_capacity_mut();
412        assert!(
413            !destination.is_empty(),
414            "a pending MAVLink frame must fit in the decoder buffer"
415        );
416        let count = self
417            .source
418            .read(destination)
419            .await
420            .map_err(|_| MessageReadError::Io)?;
421        if count == 0 {
422            return Err(MessageReadError::eof());
423        }
424        self.decoder.commit(count);
425        Ok(())
426    }
427}
mavlink_core/async_reader.rs

mavlink_core/
async_reader.rs
