mavlink_core/connection/tcp/
sync.rs

1//! TCP MAVLink connection
2
3use crate::Dialect;
4use crate::MavlinkReader;
5use crate::connection::get_socket_addr;
6use crate::connection::sync::{Connectable, DialectConnection, SyncTransport, Transport};
7use std::io;
8use std::net::{TcpListener, TcpStream, ToSocketAddrs};
9use std::sync::Mutex;
10use std::time::Duration;
11
12use super::config::{TcpConfig, TcpMode};
13
14pub fn tcpout<T: ToSocketAddrs>(address: T) -> io::Result<TcpConnection> {
15    let socket = TcpStream::connect(get_socket_addr(&address)?)?;
16    socket.set_read_timeout(Some(Duration::from_millis(100)))?;
17    connection_from_stream(socket)
18}
19
20fn connection_from_stream(socket: TcpStream) -> io::Result<TcpConnection> {
21    Ok(TcpConnection {
22        reader: Mutex::new(MavlinkReader::new(socket.try_clone()?)),
23        writer: Mutex::new(socket),
24    })
25}
26
27pub fn tcpin<T: ToSocketAddrs>(address: T) -> io::Result<TcpConnection> {
28    let listener = TcpListener::bind(get_socket_addr(&address)?)?;
29    accept(listener)
30}
31
32fn accept(listener: TcpListener) -> io::Result<TcpConnection> {
33    for incoming in listener.incoming() {
34        match incoming {
35            Ok(socket) => return connection_from_stream(socket),
36            Err(e) => println!("listener err: {e}"),
37        }
38    }
39    Err(io::Error::new(
40        io::ErrorKind::NotConnected,
41        "No incoming connections!",
42    ))
43}
44
45pub struct TcpConnection {
46    reader: Mutex<MavlinkReader<TcpStream>>,
47    writer: Mutex<TcpStream>,
48}
49
50impl SyncTransport for TcpConnection {
51    type Reader = TcpStream;
52    type Writer = TcpStream;
53
54    fn reader(&self) -> std::sync::MutexGuard<'_, MavlinkReader<Self::Reader>> {
55        self.reader.lock().unwrap()
56    }
57
58    fn lock_writer(&self) -> Option<std::sync::MutexGuard<'_, Self::Writer>> {
59        Some(self.writer.lock().unwrap())
60    }
61
62    fn set_nonblocking(
63        &self,
64        reader: &mut MavlinkReader<Self::Reader>,
65        enabled: bool,
66    ) -> io::Result<()> {
67        reader.get_mut().set_nonblocking(enabled)
68    }
69}
70
71impl TcpConfig {
72    pub(crate) fn open(&self) -> io::Result<TcpConnection> {
73        match self.mode {
74            TcpMode::TcpIn => match self.take_listener()? {
75                Some(listener) => accept(listener),
76                None => tcpin(&self.address),
77            },
78            TcpMode::TcpOut => match self.take_stream()? {
79                Some(stream) => connection_from_stream(stream),
80                None => tcpout(&self.address),
81            },
82        }
83    }
84}
85
86impl Connectable for TcpConfig {
87    fn connect_with_dialect<D: Dialect + Send + Sync>(
88        &self,
89        dialect: D,
90    ) -> io::Result<DialectConnection<D>> {
91        Ok(DialectConnection::new(
92            Transport::Tcp(self.open()?),
93            dialect,
94        ))
95    }
96}
mavlink_core/connection/tcp/sync.rs

mavlink_core/connection/tcp/
sync.rs
