msrt_uart/
frontend.rs

1//! Tokio host UART frontend.
2
3use std::io::ErrorKind;
4use std::pin::Pin;
5use std::task::{Context, Poll, Waker};
6use std::time::Instant;
7
8use msrt::endpoint::{ClientEndpoint, EndpointPoll, EngineConfig, PeerState};
9use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};
10
11use crate::event::UartFrontendEvent;
12use crate::tokio_error::{TokioError, TokioResult};
13
14const RX_BYTES: usize = 512;
15const TX_BYTES: usize = 256;
16
17/// Tokio host UART frontend adapter.
18///
19/// `T` can be a Tokio serial stream, TCP stream used as a UART-like byte stream,
20/// or any async byte stream implementing `AsyncRead + AsyncWrite + Unpin`.
21#[derive(Debug)]
22pub struct TokioUartFrontend<T> {
23    io: T,
24    endpoint: ClientEndpoint,
25    start: Instant,
26    rx_buf: [u8; RX_BYTES],
27    tx_buf: [u8; TX_BYTES],
28}
29
30impl<T> TokioUartFrontend<T>
31where
32    T: AsyncRead + AsyncWrite + Unpin,
33{
34    /// Creates a frontend adapter using default MSRT config.
35    pub fn new(io: T) -> Self {
36        Self::with_config(io, EngineConfig::default())
37    }
38
39    /// Creates a frontend adapter using `config`.
40    pub fn with_config(io: T, config: EngineConfig) -> Self {
41        Self {
42            io,
43            endpoint: ClientEndpoint::new(config),
44            start: Instant::now(),
45            rx_buf: [0; RX_BYTES],
46            tx_buf: [0; TX_BYTES],
47        }
48    }
49
50    /// Returns the endpoint peer state.
51    pub fn peer_state(&self) -> PeerState {
52        self.endpoint.peer().state()
53    }
54
55    /// Returns a shared reference to the owned async IO object.
56    pub const fn io(&self) -> &T {
57        &self.io
58    }
59
60    /// Returns a mutable reference to the owned async IO object.
61    pub fn io_mut(&mut self) -> &mut T {
62        &mut self.io
63    }
64
65    /// Consumes the adapter and returns the async IO object.
66    pub fn into_inner(self) -> T {
67        self.io
68    }
69
70    /// Starts a fresh frontend session.
71    pub fn connect(&mut self) -> TokioResult<()> {
72        self.endpoint.connect(self.now_ms())?;
73        Ok(())
74    }
75
76    /// Drops the active frontend session.
77    pub fn disconnect(&mut self) {
78        self.endpoint.disconnect();
79    }
80
81    /// Queues an application message.
82    pub fn send(&mut self, message: &[u8]) -> TokioResult<bool> {
83        Ok(self.endpoint.send(message)?.is_some())
84    }
85
86    /// Reads currently available UART bytes and feeds them into MSRT.
87    pub async fn receive_available(&mut self) -> TokioResult<usize> {
88        let mut total = 0;
89        loop {
90            let now_ms = self.now_ms();
91            let mut read_buf = ReadBuf::new(&mut self.rx_buf);
92            let waker = Waker::noop();
93            let mut context = Context::from_waker(waker);
94            match Pin::new(&mut self.io).poll_read(&mut context, &mut read_buf) {
95                Poll::Ready(Ok(())) if read_buf.filled().is_empty() => return Ok(total),
96                Poll::Ready(Ok(())) => {
97                    let n = read_buf.filled().len();
98                    total += n;
99                    let _ = self.endpoint.receive(now_ms, read_buf.filled());
100                    if n < self.rx_buf.len() {
101                        return Ok(total);
102                    }
103                }
104                Poll::Ready(Err(error)) if error.kind() == ErrorKind::WouldBlock => {
105                    return Ok(total);
106                }
107                Poll::Ready(Err(error)) if error.kind() == ErrorKind::Interrupted => continue,
108                Poll::Ready(Err(error)) => return Err(error.into()),
109                Poll::Pending => return Ok(total),
110            }
111        }
112    }
113
114    /// Polls one frontend event and writes pending UART bytes.
115    pub async fn poll(&mut self) -> TokioResult<UartFrontendEvent> {
116        let now_ms = self.now_ms();
117        loop {
118            match self.endpoint.poll(now_ms, &mut self.tx_buf)? {
119                EndpointPoll::Transmit { bytes, .. } => self.io.write_all(bytes).await?,
120                EndpointPoll::Message(message) => return Ok(UartFrontendEvent::Message(message)),
121                EndpointPoll::SendFailed(failed) => {
122                    return Ok(UartFrontendEvent::SendFailed(failed));
123                }
124                EndpointPoll::Idle => return Ok(UartFrontendEvent::Idle),
125            }
126        }
127    }
128
129    /// Runs `receive_available` followed by `poll`.
130    pub async fn tick(&mut self) -> TokioResult<UartFrontendEvent> {
131        if let Err(error) = self.receive_available().await {
132            if recoverable_transport_error(&error) {
133                return Ok(UartFrontendEvent::TransportUnavailable);
134            }
135            return Err(error);
136        }
137
138        match self.poll().await {
139            Ok(event) => Ok(event),
140            Err(error) if recoverable_transport_error(&error) => {
141                Ok(UartFrontendEvent::TransportUnavailable)
142            }
143            Err(error) => Err(error),
144        }
145    }
146
147    fn now_ms(&self) -> u64 {
148        self.start
149            .elapsed()
150            .as_millis()
151            .try_into()
152            .unwrap_or(u64::MAX)
153    }
154}
155
156fn recoverable_transport_error(error: &TokioError) -> bool {
157    let TokioError::Io(error) = error else {
158        return false;
159    };
160
161    matches!(
162        error.kind(),
163        ErrorKind::ConnectionRefused
164            | ErrorKind::ConnectionReset
165            | ErrorKind::ConnectionAborted
166            | ErrorKind::NotConnected
167            | ErrorKind::TimedOut
168    )
169}
msrt_uart/frontend.rs

msrt_uart/
frontend.rs
