mavlink_core/peek_reader.rs
1//! This module implements a buffered/peekable reader.
2//!
3//! The purpose of the buffered/peekable reader is to allow for backtracking parsers.
4//!
5//! A reader implementing the standard library's [`std::io::BufRead`] trait seems like a good fit, but
6//! it does not allow for peeking a specific number of bytes, so it provides no way to request
7//! more data from the underlying reader without consuming the existing data.
8//!
9//! This API still tries to adhere to the [`std::io::BufRead`]'s trait philosophy.
10//!
11//! The main type `PeekReader`does not implement [`std::io::Read`] itself, as there is no added benefit
12//! in doing so.
13//!
14#[cfg(all(feature = "embedded", not(feature = "std")))]
15use crate::embedded::Read;
16
17#[cfg(feature = "std")]
18use std::io::Read;
19
20#[cfg(all(doc, feature = "std"))]
21use std::io::ErrorKind;
22
23use crate::error::MessageReadError;
24
25/// A buffered/peekable reader
26///
27/// This reader wraps a type implementing [`Read`] and adds buffering via an internal buffer.
28///
29/// It allows the user to `peek` a specified number of bytes (without consuming them),
30/// to `read` bytes (consuming them), or to `consume` them after `peek`ing.
31///
32/// NOTE: This reader is generic over the size of the buffer, defaulting to MAVLink's current largest
33/// possible message size of 280 bytes
34///
35pub struct PeekReader<R, const BUFFER_SIZE: usize = 280> {
36 // Internal buffer
37 buffer: [u8; BUFFER_SIZE],
38 // The position of the next byte to read from the buffer.
39 cursor: usize,
40 // The position of the next byte to read into the buffer.
41 top: usize,
42 // The wrapped reader.
43 reader: R,
44}
45
46impl<R: Read, const BUFFER_SIZE: usize> PeekReader<R, BUFFER_SIZE> {
47 /// Instantiates a new [`PeekReader`], wrapping the provided [`Read`]er and using the default chunk size
48 pub fn new(reader: R) -> Self {
49 Self {
50 buffer: [0; BUFFER_SIZE],
51 cursor: 0,
52 top: 0,
53 reader,
54 }
55 }
56
57 /// Peeks an exact amount of bytes from the internal buffer
58 ///
59 /// If the internal buffer does not contain enough data, this function will read
60 /// from the underlying [`Read`]er until it does, an error occurs or no more data can be read (EOF).
61 ///
62 /// This function does not consume data from the buffer, so subsequent calls to `peek` or `read` functions
63 /// will still return the peeked data.
64 ///
65 /// # Errors
66 ///
67 /// - If any error occurs while reading from the underlying [`Read`]er it is returned
68 /// - If an EOF occurs and the specified amount could not be read, this function will return an [`ErrorKind::UnexpectedEof`].
69 ///
70 /// # Panics
71 ///
72 /// Will panic when attempting to read more bytes then `BUFFER_SIZE`
73 pub fn peek_exact(&mut self, amount: usize) -> Result<&[u8], MessageReadError> {
74 self.fetch(amount, false)
75 }
76
77 /// Reads a specified amount of bytes from the internal buffer
78 ///
79 /// If the internal buffer does not contain enough data, this function will read
80 /// from the underlying [`Read`]er until it does, an error occurs or no more data can be read (EOF).
81 ///
82 /// This function consumes the data from the buffer, unless an error occurs, in which case no data is consumed.
83 ///
84 /// # Errors
85 ///
86 /// - If any error occurs while reading from the underlying [`Read`]er it is returned
87 /// - If an EOF occurs and the specified amount could not be read, this function will return an [`ErrorKind::UnexpectedEof`].
88 ///
89 /// # Panics
90 ///
91 /// Will panic when attempting to read more bytes then `BUFFER_SIZE`
92 pub fn read_exact(&mut self, amount: usize) -> Result<&[u8], MessageReadError> {
93 self.fetch(amount, true)
94 }
95
96 /// Reads a byte from the internal buffer
97 ///
98 /// If the internal buffer does not contain enough data, this function will read
99 /// from the underlying [`Read`]er until it does, an error occurs or no more data can be read (EOF).
100 ///
101 /// This function consumes the data from the buffer, unless an error occurs, in which case no data is consumed.
102 ///
103 /// # Errors
104 ///
105 /// - If any error occurs while reading from the underlying [`Read`]er it is returned
106 /// - If an EOF occurs before a byte could be read, this function will return an [`ErrorKind::UnexpectedEof`].
107 ///
108 /// # Panics
109 ///
110 /// Will panic if this `PeekReader`'s `BUFFER_SIZE` is 0.
111 pub fn read_u8(&mut self) -> Result<u8, MessageReadError> {
112 let buf = self.read_exact(1)?;
113 Ok(buf[0])
114 }
115
116 /// Consumes a specified amount of bytes from the buffer
117 ///
118 /// If the internal buffer does not contain enough data, this function will consume as much data as is buffered.
119 ///
120 pub fn consume(&mut self, amount: usize) -> usize {
121 let amount = amount.min(self.top - self.cursor);
122 self.cursor += amount;
123 amount
124 }
125
126 /// Returns an immutable reference to the underlying [`Read`]er
127 ///
128 /// Reading directly from the underlying reader will cause data loss
129 pub fn reader_ref(&self) -> &R {
130 &self.reader
131 }
132
133 /// Returns a mutable reference to the underlying [`Read`]er
134 ///
135 /// Reading directly from the underlying reader will cause data loss
136 pub fn reader_mut(&mut self) -> &mut R {
137 &mut self.reader
138 }
139
140 /// Internal function to fetch data from the internal buffer and/or reader
141 fn fetch(&mut self, amount: usize, consume: bool) -> Result<&[u8], MessageReadError> {
142 assert!(BUFFER_SIZE >= amount);
143
144 loop {
145 let buffered = self.top - self.cursor;
146
147 if buffered >= amount {
148 break;
149 }
150
151 // the caller requested more bytes than we have buffered, fetch them from the reader
152 let bytes_to_read = amount - buffered;
153
154 // Check if we need to compact the buffer first
155 if self.top + bytes_to_read > BUFFER_SIZE {
156 // Move unread data to the beginning of the buffer
157 self.buffer.copy_within(self.cursor..self.top, 0);
158 self.top = buffered;
159 self.cursor = 0;
160 }
161
162 // Now we can safely read directly into the buffer
163 let end_pos = self.top + bytes_to_read;
164
165 // read needed bytes from reader
166 let bytes_read = self.reader.read(&mut self.buffer[self.top..end_pos])?;
167
168 if bytes_read == 0 {
169 return Err(MessageReadError::eof());
170 }
171
172 self.top += bytes_read;
173 }
174
175 let result = &self.buffer[self.cursor..self.cursor + amount];
176 if consume {
177 self.cursor += amount;
178 }
179 Ok(result)
180 }
181}
182
183#[cfg(test)]
184mod tests {
185 use super::*;
186
187 #[cfg(feature = "std")]
188 use std::io::Write;
189
190 #[cfg(not(feature = "std"))]
191 use embedded_io::Write;
192
193 #[test]
194 fn test_read_and_peek() {
195 let data = b"Hello, World!";
196 let mut buffer = [0u8; 280];
197
198 let mut writer: &mut [u8] = &mut buffer[..];
199 writer.write_all(data).unwrap();
200
201 let mut reader = PeekReader::<_, 280>::new(&buffer[..data.len()]);
202
203 let peeked = reader.peek_exact(5).unwrap();
204 assert_eq!(peeked, b"Hello");
205
206 let read = reader.read_exact(5).unwrap();
207 assert_eq!(read, b"Hello");
208
209 // Make sure `PeekReader::read_exact` consumed the first 5 bytes.
210 let read = reader.read_exact(8).unwrap();
211 assert_eq!(read, b", World!");
212
213 match reader.read_u8().unwrap_err() {
214 #[cfg(feature = "std")]
215 MessageReadError::Io(io_err) => {
216 assert_eq!(io_err.kind(), std::io::ErrorKind::UnexpectedEof);
217 }
218 #[cfg(not(feature = "std"))]
219 MessageReadError::Io => (),
220 _ => panic!("Expected Io error with UnexpectedEof"),
221 }
222 }
223
224 #[test]
225 #[should_panic(expected = "assertion failed")]
226 fn test_peek_exact_panics_when_amount_exceeds_buffer_size() {
227 let data = b"abcd";
228 let mut reader = PeekReader::<_, 4>::new(&data[..]);
229 let _ = reader.peek_exact(5);
230 }
231
232 #[test]
233 #[should_panic(expected = "assertion failed")]
234 fn test_read_exact_panics_when_amount_exceeds_buffer_size() {
235 let data = b"abcd";
236 let mut reader = PeekReader::<_, 4>::new(&data[..]);
237 let _ = reader.read_exact(5);
238 }
239}