mavlink_core/connection/tcp/
async.rs

1//! Async TCP MAVLink connection
2
3use std::io;
4
5use futures::lock::{Mutex, MutexGuard};
6use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
7use tokio::net::{TcpListener, TcpStream};
8
9use crate::AsyncMavlinkReader;
10use crate::Dialect;
11use crate::connection::r#async::{
12    AsyncConnectable, AsyncDialectConnection, AsyncTransport, Transport,
13};
14use crate::connection::get_socket_addr;
15use crate::connection::tcp::config::{TcpConfig, TcpMode};
16
17pub async fn tcpout<T: std::net::ToSocketAddrs>(address: T) -> io::Result<AsyncTcpConnection> {
18    let socket = TcpStream::connect(get_socket_addr(&address)?).await?;
19    Ok(AsyncTcpConnection::from_stream(socket))
20}
21
22pub async fn tcpin<T: std::net::ToSocketAddrs>(address: T) -> io::Result<AsyncTcpConnection> {
23    let listener = TcpListener::bind(get_socket_addr(&address)?).await?;
24    //For now we only accept one incoming stream: this yields until we get one
25    let (socket, _) = listener.accept().await?;
26    Ok(AsyncTcpConnection::from_stream(socket))
27}
28
29pub struct AsyncTcpConnection {
30    reader: Mutex<AsyncMavlinkReader<OwnedReadHalf>>,
31    writer: Mutex<OwnedWriteHalf>,
32}
33
34impl AsyncTcpConnection {
35    fn from_stream(socket: TcpStream) -> Self {
36        let (reader, writer) = socket.into_split();
37        Self {
38            reader: Mutex::new(AsyncMavlinkReader::new(reader)),
39            writer: Mutex::new(writer),
40        }
41    }
42}
43
44impl AsyncTransport for AsyncTcpConnection {
45    type Reader = OwnedReadHalf;
46    type Writer = OwnedWriteHalf;
47
48    fn reader(&self) -> &Mutex<AsyncMavlinkReader<Self::Reader>> {
49        &self.reader
50    }
51
52    async fn lock_writer(&self) -> Option<MutexGuard<'_, Self::Writer>> {
53        Some(self.writer.lock().await)
54    }
55}
56
57impl TcpConfig {
58    pub(crate) async fn open_async(&self) -> io::Result<AsyncTcpConnection> {
59        match self.mode {
60            TcpMode::TcpIn => match self.take_listener()? {
61                Some(listener) => {
62                    listener.set_nonblocking(true)?;
63                    let (socket, _) = TcpListener::from_std(listener)?.accept().await?;
64                    Ok(AsyncTcpConnection::from_stream(socket))
65                }
66                None => tcpin(&self.address).await,
67            },
68            TcpMode::TcpOut => match self.take_stream()? {
69                Some(stream) => {
70                    stream.set_nonblocking(true)?;
71                    Ok(AsyncTcpConnection::from_stream(TcpStream::from_std(
72                        stream,
73                    )?))
74                }
75                None => tcpout(&self.address).await,
76            },
77        }
78    }
79}
80
81impl AsyncConnectable for TcpConfig {
82    async fn connect_async_with_dialect<D>(
83        &self,
84        dialect: D,
85    ) -> io::Result<AsyncDialectConnection<D>>
86    where
87        D: Dialect + Send + Sync,
88        D::Message: Send + Sync,
89    {
90        Ok(AsyncDialectConnection::new(
91            Transport::Tcp(self.open_async().await?),
92            dialect,
93        ))
94    }
95}
mavlink_core/connection/tcp/async.rs

mavlink_core/connection/tcp/
async.rs
