mavlink_core/connection/tcp/
async.rs

1//! Async TCP MAVLink connection
2
3use std::io;
4
5use crate::async_peek_reader::AsyncPeekReader;
6use crate::connection::tcp::config::{TcpConfig, TcpMode};
7use crate::connection::{AsyncConnectable, AsyncMavConnection, get_socket_addr};
8use crate::connection_shared::{
9    ConnectionState, next_send_header, read_message_async, read_raw_message_async,
10    write_message_async, write_raw_message_async,
11};
12use crate::{MAVLinkMessageRaw, MavHeader, MavlinkVersion, Message};
13
14use async_trait::async_trait;
15use core::ops::DerefMut;
16use futures::{FutureExt, lock::Mutex};
17use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
18use tokio::net::{TcpListener, TcpStream};
19
20#[cfg(feature = "mav2-message-signing")]
21use crate::SigningConfig;
22
23pub async fn tcpout<T: std::net::ToSocketAddrs>(address: T) -> io::Result<AsyncTcpConnection> {
24    let addr = get_socket_addr(&address)?;
25
26    let socket = TcpStream::connect(addr).await?;
27
28    connection_from_stream(socket)
29}
30
31fn connection_from_stream(socket: TcpStream) -> io::Result<AsyncTcpConnection> {
32    let (reader, writer) = socket.into_split();
33
34    Ok(AsyncTcpConnection {
35        reader: Mutex::new(AsyncPeekReader::new(reader)),
36        writer: Mutex::new(TcpWrite {
37            socket: writer,
38            sequence: 0,
39        }),
40        state: ConnectionState::new(),
41    })
42}
43
44async fn accept(listener: TcpListener) -> io::Result<AsyncTcpConnection> {
45    let (socket, _) = listener.accept().await?;
46    connection_from_stream(socket)
47}
48
49pub async fn tcpin<T: std::net::ToSocketAddrs>(address: T) -> io::Result<AsyncTcpConnection> {
50    let addr = get_socket_addr(&address)?;
51    let listener = TcpListener::bind(addr).await?;
52
53    //For now we only accept one incoming stream: this yields until we get one
54    match listener.accept().await {
55        Ok((socket, _)) => {
56            let (reader, writer) = socket.into_split();
57            return Ok(AsyncTcpConnection {
58                reader: Mutex::new(AsyncPeekReader::new(reader)),
59                writer: Mutex::new(TcpWrite {
60                    socket: writer,
61                    sequence: 0,
62                }),
63                state: ConnectionState::new(),
64            });
65        }
66        Err(e) => {
67            //TODO don't println in lib
68            println!("listener err: {e}");
69        }
70    }
71    Err(io::Error::new(
72        io::ErrorKind::NotConnected,
73        "No incoming connections!",
74    ))
75}
76
77pub struct AsyncTcpConnection {
78    reader: Mutex<AsyncPeekReader<OwnedReadHalf>>,
79    writer: Mutex<TcpWrite>,
80    state: ConnectionState,
81}
82
83struct TcpWrite {
84    socket: OwnedWriteHalf,
85    sequence: u8,
86}
87
88#[async_trait::async_trait]
89impl<M: Message + Sync + Send> AsyncMavConnection<M> for AsyncTcpConnection {
90    async fn recv(&self) -> Result<(MavHeader, M), crate::error::MessageReadError> {
91        let mut reader = self.reader.lock().await;
92        read_message_async::<M, _>(reader.deref_mut(), &self.state).await
93    }
94
95    async fn recv_raw(&self) -> Result<MAVLinkMessageRaw, crate::error::MessageReadError> {
96        let mut reader = self.reader.lock().await;
97        read_raw_message_async::<M, _>(reader.deref_mut(), &self.state).await
98    }
99
100    async fn try_recv(&self) -> Result<(MavHeader, M), crate::error::MessageReadError> {
101        match self.recv().now_or_never() {
102            Some(result) => result,
103            None => Err(crate::error::MessageReadError::Io(
104                io::ErrorKind::WouldBlock.into(),
105            )),
106        }
107    }
108
109    async fn send(
110        &self,
111        header: &MavHeader,
112        data: &M,
113    ) -> Result<usize, crate::error::MessageWriteError> {
114        let mut lock = self.writer.lock().await;
115
116        let header = next_send_header(&mut lock.sequence, header);
117        write_message_async(&mut lock.socket, &self.state, header, data).await
118    }
119
120    async fn send_raw(
121        &self,
122        data: &MAVLinkMessageRaw,
123    ) -> Result<usize, crate::error::MessageWriteError> {
124        let mut lock = self.writer.lock().await;
125        write_raw_message_async(&mut lock.socket, data).await
126    }
127
128    fn set_protocol_version(&mut self, version: MavlinkVersion) {
129        self.state.set_protocol_version(version);
130    }
131
132    fn protocol_version(&self) -> MavlinkVersion {
133        self.state.protocol_version()
134    }
135
136    fn set_allow_recv_any_version(&mut self, allow: bool) {
137        self.state.set_allow_recv_any_version(allow);
138    }
139
140    fn allow_recv_any_version(&self) -> bool {
141        self.state.allow_recv_any_version()
142    }
143
144    #[cfg(feature = "mav2-message-signing")]
145    fn setup_signing(&mut self, signing_data: Option<SigningConfig>) {
146        self.state.setup_signing(signing_data);
147    }
148}
149
150#[async_trait]
151impl AsyncConnectable for TcpConfig {
152    async fn connect_async<M>(&self) -> io::Result<Box<dyn AsyncMavConnection<M> + Sync + Send>>
153    where
154        M: Message + Sync + Send,
155    {
156        let conn = match self.mode {
157            TcpMode::TcpIn => match self.take_listener()? {
158                Some(listener) => {
159                    listener.set_nonblocking(true)?;
160                    accept(TcpListener::from_std(listener)?).await
161                }
162                None => tcpin(&self.address).await,
163            },
164            TcpMode::TcpOut => match self.take_stream()? {
165                Some(stream) => {
166                    stream.set_nonblocking(true)?;
167                    connection_from_stream(TcpStream::from_std(stream)?)
168                }
169                None => tcpout(&self.address).await,
170            },
171        };
172
173        Ok(Box::new(conn?))
174    }
175}
mavlink_core/connection/tcp/async.rs

mavlink_core/connection/tcp/
async.rs
