mavlink_core/connection/
async.rs

1use std::{future::Future, io};
2
3use futures::lock::{Mutex, MutexGuard};
4use tokio::io::{AsyncRead, AsyncWrite};
5
6use crate::AsyncMavlinkReader;
7#[cfg(feature = "mav2-message-signing")]
8use crate::SigningConfig;
9use crate::error::{MessageReadError, MessageWriteError};
10use crate::static_dialect::StaticDialect;
11use crate::{
12    Dialect, MAVLinkMessageRaw, MavFrame, MavHeader, MavlinkVersion, Message,
13    connectable::ConnectionAddress,
14};
15
16/// The byte-oriented half of a Tokio transport.
17#[doc(hidden)]
18pub(crate) trait AsyncTransport: Send + Sync {
19    /// Incoming bytes.
20    type Reader: AsyncRead + Unpin + Send;
21    /// Outgoing bytes.
22    type Writer: AsyncWrite + Unpin + Send;
23
24    /// Returns the reader mutex.
25    fn reader(&self) -> &Mutex<AsyncMavlinkReader<Self::Reader>>;
26
27    /// Locks the writer. Returns `None` to skip sending when there is no writer or peer.
28    fn lock_writer(&self) -> impl Future<Output = Option<MutexGuard<'_, Self::Writer>>> + Send;
29
30    /// Whether `recv` and `recv_raw` should retry an error. Unused by `try_recv`.
31    fn retry_receive(&self, _error: &MessageReadError) -> bool {
32        false
33    }
34}
35
36/// A concrete async connection using generated messages.
37pub type AsyncConnection<M> = AsyncDialectConnection<StaticDialect<M>>;
38
39/// A concrete async connection using a dialect.
40pub struct AsyncDialectConnection<D> {
41    transport: Transport,
42    dialect: D,
43    state: crate::connection_shared::ConnectionState,
44}
45
46pub(crate) enum Transport {
47    #[cfg(feature = "transport-tcp")]
48    Tcp(super::tcp::AsyncTcpConnection),
49    #[cfg(feature = "transport-udp")]
50    Udp(super::udp::AsyncUdpConnection),
51    #[cfg(feature = "transport-direct-serial")]
52    Serial(super::direct_serial::AsyncSerialConnection),
53    File(super::file::AsyncFileConnection),
54}
55
56impl<D> AsyncDialectConnection<D> {
57    pub(crate) fn new(transport: Transport, dialect: D) -> Self {
58        Self {
59            transport,
60            dialect,
61            state: crate::connection_shared::ConnectionState::new(),
62        }
63    }
64}
65
66/// An async MAVLink connection
67pub trait AsyncMavConnection<M: Sync + Send>: Send + Sync {
68    /// Receive a mavlink message.
69    ///
70    /// Yield until a valid frame is received, ignoring invalid messages.
71    fn recv(&self) -> impl Future<Output = Result<(MavHeader, M), MessageReadError>> + Send;
72
73    /// Receive a raw, unparsed mavlink message.
74    ///
75    /// Yield until a valid frame is received, ignoring invalid messages.
76    fn recv_raw(&self) -> impl Future<Output = Result<MAVLinkMessageRaw, MessageReadError>> + Send;
77
78    /// Receive a MAVLink message without retrying read or parse errors.
79    ///
80    /// Waits for the reader lock and incoming data.
81    fn try_recv(&self) -> impl Future<Output = Result<(MavHeader, M), MessageReadError>> + Send;
82
83    /// Send a mavlink message
84    fn send(
85        &self,
86        header: &MavHeader,
87        data: &M,
88    ) -> impl Future<Output = Result<usize, MessageWriteError>> + Send;
89
90    /// Send a raw, unparsed mavlink message
91    fn send_raw(
92        &self,
93        data: &MAVLinkMessageRaw,
94    ) -> impl Future<Output = Result<usize, MessageWriteError>> + Send;
95
96    /// Sets the MAVLink version to use for receiving (when `allow_recv_any_version()` is `false`) and sending messages.
97    fn set_protocol_version(&mut self, version: MavlinkVersion);
98    /// Gets the currently used MAVLink version
99    fn protocol_version(&self) -> MavlinkVersion;
100
101    /// Set wether MAVLink messages of either version may be received.
102    ///
103    /// If set to false only messages of the version configured with `set_protocol_version()` are received.
104    fn set_allow_recv_any_version(&mut self, allow: bool);
105
106    /// Wether messages of any MAVLink version may be received.
107    fn allow_recv_any_version(&self) -> bool;
108
109    /// Write whole frame.
110    fn send_frame(
111        &self,
112        frame: &MavFrame<M>,
113    ) -> impl Future<Output = Result<usize, MessageWriteError>> + Send
114    where
115        M: Message,
116    {
117        self.send(&frame.header, &frame.msg)
118    }
119
120    /// Read whole frame.
121    fn recv_frame(&self) -> impl Future<Output = Result<MavFrame<M>, MessageReadError>> + Send
122    where
123        M: Message,
124    {
125        async move {
126            let (header, msg) = self.recv().await?;
127            Ok(MavFrame {
128                header,
129                msg,
130                protocol_version: self.protocol_version(),
131            })
132        }
133    }
134
135    /// Send a message with default header.
136    fn send_default(
137        &self,
138        data: &M,
139    ) -> impl Future<Output = Result<usize, MessageWriteError>> + Send {
140        async move { self.send(&MavHeader::default(), data).await }
141    }
142
143    /// Setup secret key used for message signing, or disable message signing.
144    #[cfg(feature = "mav2-message-signing")]
145    fn setup_signing(&mut self, signing_data: Option<SigningConfig>);
146}
147
148impl<D: Dialect + Send + Sync> AsyncMavConnection<D::Message> for AsyncDialectConnection<D>
149where
150    D::Message: Send + Sync,
151{
152    async fn recv(&self) -> Result<(MavHeader, D::Message), MessageReadError> {
153        dispatch_transport!(&self.transport, |transport| {
154            let mut reader = transport.reader().lock().await;
155            loop {
156                match crate::connection_shared::read_message_async_with_dialect(
157                    &mut reader,
158                    &self.state,
159                    &self.dialect,
160                )
161                .await
162                {
163                    Ok(message) => return Ok(message),
164                    Err(error) if transport.retry_receive(&error) => {}
165                    Err(error) => return Err(error),
166                }
167            }
168        })
169    }
170
171    async fn recv_raw(&self) -> Result<MAVLinkMessageRaw, MessageReadError> {
172        dispatch_transport!(&self.transport, |transport| {
173            let mut reader = transport.reader().lock().await;
174            loop {
175                match crate::connection_shared::read_raw_message_async_with_dialect(
176                    &mut reader,
177                    &self.state,
178                    &self.dialect,
179                )
180                .await
181                {
182                    Ok(message) => return Ok(message),
183                    Err(error) if transport.retry_receive(&error) => {}
184                    Err(error) => return Err(error),
185                }
186            }
187        })
188    }
189
190    async fn try_recv(&self) -> Result<(MavHeader, D::Message), MessageReadError> {
191        dispatch_transport!(&self.transport, |transport| {
192            let mut reader = transport.reader().lock().await;
193            crate::connection_shared::read_message_async_with_dialect(
194                &mut reader,
195                &self.state,
196                &self.dialect,
197            )
198            .await
199        })
200    }
201
202    async fn send(
203        &self,
204        header: &MavHeader,
205        data: &D::Message,
206    ) -> Result<usize, MessageWriteError> {
207        dispatch_transport!(&self.transport, |transport| {
208            let Some(mut writer) = transport.lock_writer().await else {
209                return Ok(0);
210            };
211            let header = self.state.next_header(header);
212            crate::connection_shared::write_message_async_with_dialect(
213                &mut *writer,
214                &self.state,
215                header,
216                &self.dialect,
217                data,
218            )
219            .await
220        })
221    }
222
223    async fn send_raw(&self, data: &MAVLinkMessageRaw) -> Result<usize, MessageWriteError> {
224        dispatch_transport!(&self.transport, |transport| {
225            let Some(mut writer) = transport.lock_writer().await else {
226                return Ok(0);
227            };
228            crate::connection_shared::write_raw_message_async(&mut *writer, data).await
229        })
230    }
231
232    fn set_protocol_version(&mut self, version: MavlinkVersion) {
233        self.state.set_protocol_version(version);
234    }
235
236    fn protocol_version(&self) -> MavlinkVersion {
237        self.state.protocol_version()
238    }
239
240    fn set_allow_recv_any_version(&mut self, allow: bool) {
241        self.state.set_allow_recv_any_version(allow);
242    }
243
244    fn allow_recv_any_version(&self) -> bool {
245        self.state.allow_recv_any_version()
246    }
247
248    #[cfg(feature = "mav2-message-signing")]
249    fn setup_signing(&mut self, signing_data: Option<SigningConfig>) {
250        self.state.setup_signing(signing_data);
251    }
252}
253
254/// Connect asynchronously to a MAVLink node by address string.
255///
256/// The address must be in one of the following formats:
257///
258///  * `tcpin:<addr>:<port>` to create a TCP server, listening for an incoming connection
259///  * `tcpout:<addr>:<port>` to create a TCP client
260///  * `udpin:<addr>:<port>` to create a UDP server, listening for incoming packets
261///  * `udpout:<addr>:<port>` to create a UDP client
262///  * `udpbcast:<addr>:<port>` to create a UDP broadcast
263///  * `serial:<port>:<baudrate>` to create a serial connection
264///  * `file:<path>` to extract file data, writing to such a connection does nothing
265///
266/// The address selects the transport stored in the returned connection.
267///
268/// # Errors
269///
270/// - [`AddrNotAvailable`] if the address string could not be parsed as a valid MAVLink address
271/// - When the connection could not be established a corresponding [`io::Error`] is returned
272///
273/// [`AddrNotAvailable`]: io::ErrorKind::AddrNotAvailable
274pub async fn connect_async<M: Message + Sync + Send>(
275    address: &str,
276) -> io::Result<AsyncConnection<M>> {
277    ConnectionAddress::parse_address(address)?
278        .connect_async()
279        .await
280}
281
282/// Connect asynchronously to a MAVLink node using a runtime-loaded dialect.
283///
284/// The accepted address formats and errors are the same as [`connect_async`].
285pub async fn connect_async_with_dialect<D: Dialect + Send + Sync>(
286    address: &str,
287    dialect: D,
288) -> io::Result<AsyncDialectConnection<D>>
289where
290    D::Message: Send + Sync,
291{
292    ConnectionAddress::parse_address(address)?
293        .connect_async_with_dialect(dialect)
294        .await
295}
296
297/// A MAVLink connection address that can be connected to, establishing an [`AsyncMavConnection`]
298///
299/// This is the `async` version of `Connectable`.
300pub trait AsyncConnectable: Sync {
301    /// Attempt to establish an asynchronous MAVLink connection
302    fn connect_async<M: Message + Sync + Send>(
303        &self,
304    ) -> impl Future<Output = io::Result<AsyncConnection<M>>> + Send {
305        self.connect_async_with_dialect(StaticDialect::<M>::new())
306    }
307
308    /// Attempt to establish an asynchronous MAVLink connection with a runtime-loaded dialect.
309    fn connect_async_with_dialect<D: Dialect + Send + Sync>(
310        &self,
311        dialect: D,
312    ) -> impl Future<Output = io::Result<AsyncDialectConnection<D>>> + Send
313    where
314        D::Message: Send + Sync;
315}
316
317impl AsyncConnectable for ConnectionAddress {
318    async fn connect_async_with_dialect<D: Dialect + Send + Sync>(
319        &self,
320        dialect: D,
321    ) -> io::Result<AsyncDialectConnection<D>>
322    where
323        D::Message: Send + Sync,
324    {
325        match self {
326            #[cfg(feature = "transport-tcp")]
327            Self::Tcp(c) => c.connect_async_with_dialect(dialect).await,
328            #[cfg(feature = "transport-udp")]
329            Self::Udp(c) => c.connect_async_with_dialect(dialect).await,
330            #[cfg(feature = "transport-direct-serial")]
331            Self::Serial(c) => c.connect_async_with_dialect(dialect).await,
332            Self::File(c) => c.connect_async_with_dialect(dialect).await,
333        }
334    }
335}
mavlink_core/connection/async.rs

mavlink_core/connection/
async.rs
