deser_core/stream/buffer.rs
1use alloc::vec;
2use alloc::vec::Vec;
3
4use crate::Position;
5use crate::de::{Deserialize, DeserializeDriver, Frame, Progress, StreamDeserializer};
6use crate::error::{Error, ErrorKind};
7
8/// The minimum number of bytes offered to read into.
9const READ_SIZE: usize = 8 * 1024;
10
11/// The state of an [`InputBuffer`], see [`InputBuffer::poll`].
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
13pub enum Status {
14 /// A value is complete and can be deserialized.
15 Ready,
16 /// More input is needed.
17 NeedInput,
18 /// There are no more values.
19 End,
20}
21
22/// Splits a stream into values without doing IO.
23///
24/// The buffer holds the data of a stream that was read so far and splits it
25/// into values with a [`StreamDeserializer`]. It does not do IO itself
26/// which makes it usable with any kind of IO: [`poll`](Self::poll) reports
27/// if a value is ready or if more input is needed. Input is read into
28/// [`read_buf`](Self::read_buf) and committed with
29/// [`filled`](Self::filled) (or [`set_eof`](Self::set_eof) at the end of
30/// the stream). Once a value is ready it's deserialized with
31/// [`deserialize`](Self::deserialize):
32///
33/// ```
34/// # use deser::de::{DeserializeDriver, Frame, StreamDeserializer};
35/// # use deser::Error;
36/// # struct Lines;
37/// # impl StreamDeserializer for Lines {
38/// # fn frame(&mut self, input: &[u8], eof: bool) -> Result<Frame, Error> {
39/// # Ok(match input.iter().position(|&b| b == b'\n') {
40/// # Some(end) => Frame::Value { start: 0, end, consumed: end + 1 },
41/// # None if eof && input.is_empty() => Frame::End,
42/// # None if eof => Frame::Value { start: 0, end: input.len(), consumed: input.len() },
43/// # None => Frame::Incomplete { consumed: 0 },
44/// # })
45/// # }
46/// # fn drive_frame<'de>(&mut self, frame: &'de [u8], driver: &mut DeserializeDriver<'_, 'de>) -> Result<(), Error> {
47/// # let value: u64 = std::str::from_utf8(frame).unwrap().parse().unwrap();
48/// # driver.emit(value)
49/// # }
50/// # }
51/// use std::io::Read;
52/// use deser::stream::{InputBuffer, Status};
53///
54/// fn read_all(mut input: impl Read) -> Result<Vec<u64>, deser::Error> {
55/// // `Lines` is the stream deserializer of a format with a number
56/// // per line
57/// let mut buffer = InputBuffer::new(Lines);
58/// let mut values = Vec::new();
59/// loop {
60/// match buffer.poll()? {
61/// Status::Ready => values.push(buffer.deserialize()?),
62/// Status::End => return Ok(values),
63/// Status::NeedInput => match input.read(buffer.read_buf())? {
64/// 0 => buffer.set_eof(),
65/// read => buffer.filled(read),
66/// },
67/// }
68/// }
69/// }
70///
71/// assert_eq!(read_all(&b"1\n2\n3"[..]).unwrap(), [1, 2, 3]);
72/// ```
73///
74/// The offsets, lines and columns of errors refer to the stream.
75///
76/// Stream deserializers which support it can also deserialize values while
77/// their input arrives, see [`feed`](Self::feed).
78pub struct InputBuffer<D: StreamDeserializer> {
79 deserializer: D,
80 // `data[start..end]` holds the input that was not consumed yet, the
81 // data after `end` is space to read into.
82 data: Vec<u8>,
83 start: usize,
84 end: usize,
85 eof: bool,
86 // the position of `data[start]` in the stream
87 position: Position,
88 // the frame of the value which is ready (relative to `start`)
89 ready: Option<(usize, usize, usize)>,
90 // `true` once the deserializer reported the end or failed
91 done: bool,
92 failed: bool,
93 // a value is being fed into a driver
94 feeding: bool,
95}
96
97impl<D: StreamDeserializer> InputBuffer<D> {
98 /// Creates an empty buffer.
99 ///
100 /// To continue a stream whose context is known (for instance the
101 /// names of the columns of a CSV file), create the stream deserializer
102 /// with that context.
103 pub fn new(deserializer: D) -> InputBuffer<D> {
104 InputBuffer {
105 deserializer,
106 data: Vec::new(),
107 start: 0,
108 end: 0,
109 eof: false,
110 position: Position::start(),
111 ready: None,
112 done: false,
113 failed: false,
114 feeding: false,
115 }
116 }
117
118 /// Returns the stream deserializer.
119 pub fn deserializer(&self) -> &D {
120 &self.deserializer
121 }
122
123 /// Returns the stream deserializer and the input that was read but
124 /// not consumed.
125 pub fn into_parts(mut self) -> (D, Vec<u8>) {
126 self.data.truncate(self.end);
127 self.data.drain(..self.start);
128 (self.deserializer, self.data)
129 }
130
131 /// Returns the number of bytes of the stream that were consumed.
132 ///
133 /// This is the offset of the unconsumed input in the stream.
134 pub fn offset(&self) -> usize {
135 self.position.offset
136 }
137
138 /// Returns the number of bytes which were read but not consumed.
139 pub fn buffered(&self) -> usize {
140 self.end - self.start
141 }
142
143 /// Returns `true` if the end of the stream was reached.
144 pub fn is_eof(&self) -> bool {
145 self.eof
146 }
147
148 /// Discards the first bytes of the unconsumed input.
149 fn consume(&mut self, len: usize) {
150 self.position
151 .advance(&self.data[self.start..self.start + len]);
152 self.start += len;
153 }
154
155 /// Checks if the next value is ready.
156 ///
157 /// This invokes the stream deserializer to find the next value if
158 /// needed. Once the
159 /// status is [`Status::Ready`], the value has to be deserialized with
160 /// [`deserialize`](Self::deserialize) before the next one can be found.
161 /// If the stream deserializer fails, all further calls fail.
162 pub fn poll(&mut self) -> Result<Status, Error> {
163 if self.ready.is_some() {
164 return Ok(Status::Ready);
165 }
166 if self.failed {
167 return Err(failed_error());
168 }
169 if self.feeding {
170 return Err(Error::new(ErrorKind::Unexpected, "a value is being fed"));
171 }
172 if self.done {
173 return Ok(Status::End);
174 }
175 loop {
176 let input = &self.data[self.start..self.end];
177 let frame = match self.deserializer.frame(input, self.eof) {
178 Ok(frame) => frame,
179 Err(err) => {
180 self.failed = true;
181 let base = self.position;
182 let err = if self.deserializer.is_text() {
183 err.resolve_position(input)
184 } else {
185 err
186 };
187 return Err(err.shift_position(base));
188 }
189 };
190 match frame {
191 Frame::Value {
192 start,
193 end,
194 consumed,
195 } => {
196 assert!(
197 start <= end && end <= consumed && consumed <= input.len(),
198 "invalid frame"
199 );
200 self.ready = Some((start, end, consumed));
201 return Ok(Status::Ready);
202 }
203 Frame::Incomplete { consumed } => {
204 assert!(consumed <= input.len(), "invalid frame");
205 // a value might follow the discarded data
206 if consumed > 0 {
207 self.consume(consumed);
208 continue;
209 }
210 if self.eof {
211 // the deserializer cannot get more input
212 self.failed = true;
213 return Err(Error::new(ErrorKind::EndOfFile, "unexpected end of input")
214 .shift_position(self.position));
215 }
216 return Ok(Status::NeedInput);
217 }
218 Frame::End => {
219 assert!(self.eof, "end of values before the end of the input");
220 self.done = true;
221 return Ok(Status::End);
222 }
223 }
224 }
225 }
226
227 /// Checks if another value follows.
228 ///
229 /// Returns [`Status::Ready`] if a value follows (it does not need to be
230 /// complete), [`Status::End`] if there are no more values and
231 /// [`Status::NeedInput`] if more input is needed to know. The value is
232 /// then read with [`feed`](Self::feed) or, once
233 /// [`poll`](Self::poll) reports it's complete, with
234 /// [`deserialize`](Self::deserialize). If the stream deserializer
235 /// cannot find the start of a value on its own (see
236 /// [`StreamDeserializer::peek`]), the value is framed which means that
237 /// it's buffered completely.
238 pub fn peek(&mut self) -> Result<Status, Error> {
239 if self.ready.is_some() || self.feeding {
240 return Ok(Status::Ready);
241 }
242 if self.failed {
243 return Err(failed_error());
244 }
245 if self.done {
246 return Ok(Status::End);
247 }
248 loop {
249 let input = &self.data[self.start..self.end];
250 let progress = match self.deserializer.peek(input, self.eof) {
251 Ok(Some(progress)) => progress,
252 Ok(None) => return self.poll(),
253 Err(err) => {
254 self.failed = true;
255 let err = if self.deserializer.is_text() {
256 err.resolve_position(input)
257 } else {
258 err
259 };
260 return Err(err.shift_position(self.position));
261 }
262 };
263 match progress {
264 Progress::Done { consumed } => {
265 assert!(consumed <= input.len(), "invalid progress");
266 self.consume(consumed);
267 return Ok(Status::Ready);
268 }
269 Progress::NeedMore { consumed } => {
270 assert!(consumed <= input.len(), "invalid progress");
271 if consumed > 0 {
272 self.consume(consumed);
273 continue;
274 }
275 if self.eof {
276 self.failed = true;
277 return Err(Error::new(ErrorKind::EndOfFile, "unexpected end of input")
278 .shift_position(self.position));
279 }
280 return Ok(Status::NeedInput);
281 }
282 Progress::End => {
283 assert!(self.eof, "end of values before the end of the input");
284 self.done = true;
285 return Ok(Status::End);
286 }
287 }
288 }
289 }
290
291 /// Returns `true` if the stream deserializer can deserialize values
292 /// while their input arrives.
293 ///
294 /// See [`StreamDeserializer::supports_feed`] and [`feed`](Self::feed).
295 pub fn supports_feed(&self) -> bool {
296 self.deserializer.supports_feed()
297 }
298
299 /// Feeds the input into the driver of the next value.
300 ///
301 /// This is the alternative to [`poll`](Self::poll) and
302 /// [`deserialize`](Self::deserialize) for stream deserializers which
303 /// support it (see [`supports_feed`](Self::supports_feed)) and values
304 /// which do not borrow from the input. If the value was framed already
305 /// (by [`peek`](Self::peek) of a format that cannot find the start of a
306 /// value otherwise), it's deserialized from its frame. The input is fed into the driver until the
307 /// value is complete ([`Status::Ready`]), the input is consumed as it's
308 /// used. If more input is needed ([`Status::NeedInput`]) the method has
309 /// to be invoked again with the same driver once more input was read.
310 /// In the meantime the buffer cannot be used otherwise. After an error
311 /// the value is abandoned, whether the stream can continue with the next
312 /// value depends on the stream deserializer.
313 ///
314 /// ```
315 /// # use deser::de::{Frame, Progress, StreamDeserializer};
316 /// # use deser::Error;
317 /// # /// A format with sequences of digits (without separators).
318 /// # #[derive(Default)]
319 /// # struct Digits { started: bool }
320 /// # impl StreamDeserializer for Digits {
321 /// # fn frame(&mut self, _: &[u8], _: bool) -> Result<Frame, Error> { unimplemented!() }
322 /// # fn drive_frame<'de>(&mut self, _: &'de [u8], _: &mut DeserializeDriver<'_, 'de>) -> Result<(), Error> { unimplemented!() }
323 /// # fn supports_feed(&self) -> bool { true }
324 /// # fn feed(&mut self, input: &[u8], _: usize, eof: bool, driver: &mut DeserializeDriver<'_, '_>) -> Result<Progress, Error> {
325 /// # if !self.started {
326 /// # if input.is_empty() && eof { return Ok(Progress::End); }
327 /// # driver.emit(deser::Event::seq_start())?;
328 /// # self.started = true;
329 /// # }
330 /// # for digit in input { driver.emit(u64::from(digit - b'0'))?; }
331 /// # if eof {
332 /// # driver.emit(deser::Event::SeqEnd)?;
333 /// # self.started = false;
334 /// # return Ok(Progress::Done { consumed: input.len() });
335 /// # }
336 /// # Ok(Progress::NeedMore { consumed: input.len() })
337 /// # }
338 /// # }
339 /// use deser::de::DeserializeDriver;
340 /// use deser::stream::{InputBuffer, Status};
341 ///
342 /// // `Digits` is the stream deserializer of a format with a sequence
343 /// // of digits
344 /// let mut buffer = InputBuffer::new(Digits::default());
345 /// let mut out = None::<Vec<u32>>;
346 /// {
347 /// let mut driver = DeserializeDriver::new(&mut out);
348 /// for chunk in [&b"12"[..], b"3"] {
349 /// buffer.extend_from_slice(chunk);
350 /// assert_eq!(buffer.feed(&mut driver).unwrap(), Status::NeedInput);
351 /// }
352 /// buffer.set_eof();
353 /// assert_eq!(buffer.feed(&mut driver).unwrap(), Status::Ready);
354 /// }
355 /// assert_eq!(out.unwrap(), [1, 2, 3]);
356 /// ```
357 ///
358 /// # Panics
359 ///
360 /// Panics if the stream deserializer does not support feeding.
361 pub fn feed(&mut self, driver: &mut DeserializeDriver<'_, '_>) -> Result<Status, Error> {
362 assert!(
363 self.deserializer.supports_feed(),
364 "the stream deserializer does not support feeding"
365 );
366 // a value that was framed already (see `peek`)
367 if self.ready.is_some() {
368 return self.drive_transient(driver).map(|()| Status::Ready);
369 }
370 if self.failed {
371 return Err(failed_error());
372 }
373 if self.done {
374 return Ok(Status::End);
375 }
376 let input = &self.data[self.start..self.end];
377 let rv = self
378 .deserializer
379 .feed(input, self.position.offset, self.eof, driver);
380 match rv {
381 Ok(Progress::Done { consumed }) => {
382 assert!(consumed <= input.len(), "invalid progress");
383 self.consume(consumed);
384 self.feeding = false;
385 Ok(Status::Ready)
386 }
387 Ok(Progress::NeedMore { consumed }) => {
388 assert!(consumed <= input.len(), "invalid progress");
389 self.consume(consumed);
390 self.feeding = true;
391 if self.eof {
392 self.failed = true;
393 return Err(self.locate(
394 Error::new(ErrorKind::EndOfFile, "unexpected end of input")
395 .with_offset(self.position.offset),
396 ));
397 }
398 Ok(Status::NeedInput)
399 }
400 Ok(Progress::End) => {
401 assert!(self.eof, "end of values before the end of the input");
402 self.done = true;
403 self.feeding = false;
404 Ok(Status::End)
405 }
406 // whether the stream can continue is up to the deserializer
407 Err(err) => {
408 self.feeding = false;
409 Err(self.locate(err))
410 }
411 }
412 }
413
414 /// Resolves the line and column of an error with an offset in the
415 /// stream.
416 ///
417 /// This is only possible for offsets in the buffered data.
418 fn locate(&self, err: Error) -> Error {
419 err.map_each(|err| match err.offset() {
420 Some(offset)
421 if self.deserializer.is_text()
422 && err.line().is_none()
423 && offset >= self.position.offset
424 && offset - self.position.offset <= self.end - self.start =>
425 {
426 let mut position = self.position;
427 position
428 .advance(&self.data[self.start..self.start + offset - self.position.offset]);
429 err.with_position(offset, position.line, position.column)
430 }
431 _ => err,
432 })
433 }
434
435 /// Returns the buffer to read the next input into.
436 ///
437 /// After data was placed in the buffer, [`filled`](Self::filled) has
438 /// to be called with its length. The buffer is never empty.
439 pub fn read_buf(&mut self) -> &mut [u8] {
440 if self.data.len() - self.end < READ_SIZE {
441 // move the unconsumed input to the front before growing
442 if self.start > 0 {
443 self.data.copy_within(self.start..self.end, 0);
444 self.end -= self.start;
445 self.start = 0;
446 }
447 if self.data.len() - self.end < READ_SIZE {
448 let len = (self.end + READ_SIZE).max(self.data.len() * 2);
449 // a zeroed allocation instead of resizing, which writes the
450 // zeroes one by one without optimizations (and in miri)
451 let mut data = vec![0; len];
452 data[..self.end].copy_from_slice(&self.data[..self.end]);
453 self.data = data;
454 }
455 }
456 &mut self.data[self.end..]
457 }
458
459 /// Adds data that was read into [`read_buf`](Self::read_buf).
460 ///
461 /// # Panics
462 ///
463 /// Panics if the length exceeds the buffer or if the end of the stream
464 /// was reached.
465 pub fn filled(&mut self, len: usize) {
466 assert!(!self.eof, "data after the end of the stream");
467 assert!(
468 len <= self.data.len() - self.end,
469 "more data than read into"
470 );
471 self.end += len;
472 }
473
474 /// Marks the end of the stream.
475 pub fn set_eof(&mut self) {
476 self.eof = true;
477 }
478
479 /// Adds input by copying it into the buffer.
480 ///
481 /// This is an alternative to [`read_buf`](Self::read_buf) and
482 /// [`filled`](Self::filled) for input that is already in memory.
483 pub fn extend_from_slice(&mut self, mut input: &[u8]) {
484 while !input.is_empty() {
485 let buf = self.read_buf();
486 let len = buf.len().min(input.len());
487 buf[..len].copy_from_slice(&input[..len]);
488 self.filled(len);
489 input = &input[len..];
490 }
491 }
492
493 /// Takes the frame of the ready value.
494 ///
495 /// Returns the range of the frame in the data and its position.
496 fn take_ready(&mut self) -> (core::ops::Range<usize>, Position) {
497 let (start, end, consumed) = self
498 .ready
499 .take()
500 .expect("no value is ready, poll the buffer first");
501 let mut position = self.position;
502 position.advance(&self.data[self.start..self.start + start]);
503 let range = self.start + start..self.start + end;
504 self.consume(consumed);
505 (range, position)
506 }
507
508 /// Deserializes the ready value.
509 ///
510 /// The value can borrow from the buffer.
511 ///
512 /// # Panics
513 ///
514 /// Panics if no value is ready (see [`poll`](Self::poll)).
515 pub fn deserialize<'a, T: Deserialize<'a>>(&'a mut self) -> Result<T, Error> {
516 self.deserialize_with(|_| {})
517 }
518
519 /// Deserializes the ready value with a configured driver.
520 ///
521 /// The callback is invoked with the driver before the value is
522 /// deserialized, for instance to add [`Layer`](crate::de::Layer)s.
523 ///
524 /// # Panics
525 ///
526 /// Panics if no value is ready (see [`poll`](Self::poll)).
527 pub fn deserialize_with<'a, T, F>(&'a mut self, setup: F) -> Result<T, Error>
528 where
529 T: Deserialize<'a>,
530 F: FnOnce(&mut DeserializeDriver<'_, 'a>),
531 {
532 let mut out = None;
533 {
534 let mut driver = DeserializeDriver::new(&mut out);
535 setup(&mut driver);
536 self.drive(&mut driver)?;
537 }
538 out.ok_or_else(|| Error::new(ErrorKind::EndOfFile, "empty input"))
539 }
540
541 /// Feeds the events of the ready value into a driver.
542 ///
543 /// This is useful to deserialize into a custom
544 /// [`Sink`](crate::de::Sink).
545 ///
546 /// # Panics
547 ///
548 /// Panics if no value is ready (see [`poll`](Self::poll)).
549 pub fn drive<'a>(&'a mut self, driver: &mut DeserializeDriver<'_, 'a>) -> Result<(), Error> {
550 let (range, position) = self.take_ready();
551 let frame = &self.data[range];
552 self.deserializer
553 .drive_frame(frame, driver)
554 .map_err(|err| err.shift_position(position))
555 }
556
557 /// Feeds the events of the ready value into a driver for any lifetime.
558 ///
559 /// This is like [`drive`](Self::drive) but the value cannot borrow
560 /// from the buffer: the driver is lent out with
561 /// [`DeserializeDriver::transient`], borrowed data is passed on like
562 /// data that is only valid for the call. This allows driving a value
563 /// into a driver which outlives the buffer's data, for instance to
564 /// implement [`Deserializer`](crate::de::Deserializer) for a reader.
565 ///
566 /// # Panics
567 ///
568 /// Panics if no value is ready (see [`poll`](Self::poll)).
569 pub fn drive_transient(&mut self, driver: &mut DeserializeDriver<'_, '_>) -> Result<(), Error> {
570 let (range, position) = self.take_ready();
571 let frame = &self.data[range];
572 let deserializer = &mut self.deserializer;
573 driver
574 .transient(|driver| deserializer.drive_frame(frame, driver))
575 .map_err(|err| err.shift_position(position))
576 }
577
578 /// Creates the error for a value where none is expected.
579 ///
580 /// The error refers to the start of the ready value. Adapters use
581 /// this to check that a stream ends after a value (see
582 /// `Reader::end` of `deser::io`).
583 ///
584 /// # Panics
585 ///
586 /// Panics if no value is ready (see [`poll`](Self::poll)).
587 pub fn trailing_error(&self) -> Error {
588 let (start, _, _) = self.ready.expect("no value is ready");
589 let mut position = self.position;
590 position.advance(&self.data[self.start..self.start + start]);
591 Error::new(ErrorKind::Unexpected, "unexpected value after the end")
592 .with_position(0, 1, 1)
593 .shift_position(position)
594 }
595}
596
597#[cold]
598fn failed_error() -> Error {
599 Error::new(ErrorKind::Unexpected, "cannot continue after an error")
600}