deser_core/stream/buffer.rs
1use alloc::vec;
2use alloc::vec::Vec;
3
4use crate::de::{Deserialize, DeserializeDriver, Frame, Progress, StreamDeserializer};
5use crate::error::{Error, ErrorKind};
6use crate::{Context, Position};
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 [`drive_partial`](Self::drive_partial).
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 deserialized in parts
94 partial: bool,
95 // the context of the values (see `set_context`)
96 context: Context,
97}
98
99impl<D: StreamDeserializer> InputBuffer<D> {
100 /// Creates an empty buffer.
101 ///
102 /// To continue a stream whose context is known (for instance the
103 /// names of the columns of a CSV file), create the stream deserializer
104 /// with that context.
105 ///
106 /// The values are deserialized in the context of the deserializer (see
107 /// [`StreamDeserializer::context`]) unless
108 /// [`set_context`](Self::set_context) replaces it.
109 pub fn new(deserializer: D) -> InputBuffer<D> {
110 let context = deserializer.context();
111 InputBuffer {
112 deserializer,
113 data: Vec::new(),
114 start: 0,
115 end: 0,
116 eof: false,
117 position: Position::start(),
118 ready: None,
119 done: false,
120 failed: false,
121 partial: false,
122 context,
123 }
124 }
125
126 /// Sets the context the values are deserialized in.
127 ///
128 /// This replaces the context of the deserializer. The context is
129 /// given to the drivers the values are deserialized with (see
130 /// [`DeserializeDriver::set_default_context`]): a context the driver
131 /// has already takes precedence for the types it has a value for.
132 pub fn set_context(&mut self, context: Context) {
133 self.context = context;
134 }
135
136 /// Returns the context the values are deserialized in.
137 pub fn context(&self) -> &Context {
138 &self.context
139 }
140
141 /// Gives the context to a driver (for the types its context has no
142 /// value for).
143 #[inline]
144 fn apply_context(&self, driver: &mut DeserializeDriver<'_, '_>) {
145 if !self.context.is_empty() {
146 driver.set_default_context(self.context.clone());
147 }
148 }
149
150 /// Returns the stream deserializer.
151 pub fn deserializer(&self) -> &D {
152 &self.deserializer
153 }
154
155 /// Returns the stream deserializer and the input that was read but
156 /// not consumed.
157 pub fn into_parts(mut self) -> (D, Vec<u8>) {
158 self.data.truncate(self.end);
159 self.data.drain(..self.start);
160 (self.deserializer, self.data)
161 }
162
163 /// Returns the number of bytes of the stream that were consumed.
164 ///
165 /// This is the offset of the unconsumed input in the stream.
166 pub fn offset(&self) -> usize {
167 self.position.offset
168 }
169
170 /// Returns the number of bytes which were read but not consumed.
171 pub fn buffered(&self) -> usize {
172 self.end - self.start
173 }
174
175 /// Returns `true` if the end of the stream was reached.
176 pub fn is_eof(&self) -> bool {
177 self.eof
178 }
179
180 /// Discards the first bytes of the unconsumed input.
181 fn consume(&mut self, len: usize) {
182 self.position
183 .advance(&self.data[self.start..self.start + len]);
184 self.start += len;
185 }
186
187 /// Checks if the next value is ready.
188 ///
189 /// This invokes the stream deserializer to find the next value if
190 /// needed. Once the
191 /// status is [`Status::Ready`], the value has to be deserialized with
192 /// [`deserialize`](Self::deserialize) before the next one can be found.
193 /// If the stream deserializer fails, all further calls fail.
194 pub fn poll(&mut self) -> Result<Status, Error> {
195 if self.ready.is_some() {
196 return Ok(Status::Ready);
197 }
198 if self.failed {
199 return Err(failed_error());
200 }
201 if self.partial {
202 return Err(Error::new(ErrorKind::InvalidState, "a value is being fed"));
203 }
204 if self.done {
205 return Ok(Status::End);
206 }
207 loop {
208 let input = &self.data[self.start..self.end];
209 let frame = match self.deserializer.frame(input, self.eof) {
210 Ok(frame) => frame,
211 Err(mut err) => {
212 self.failed = true;
213 let base = self.position;
214 if self.deserializer.is_text() {
215 err.resolve_position(input);
216 }
217 return Err(err.shift_position(base));
218 }
219 };
220 match frame {
221 Frame::Value {
222 start,
223 end,
224 consumed,
225 } => {
226 assert!(
227 start <= end && end <= consumed && consumed <= input.len(),
228 "invalid frame"
229 );
230 self.ready = Some((start, end, consumed));
231 return Ok(Status::Ready);
232 }
233 Frame::Incomplete { consumed } => {
234 assert!(consumed <= input.len(), "invalid frame");
235 // a value might follow the discarded data
236 if consumed > 0 {
237 self.consume(consumed);
238 continue;
239 }
240 if self.eof {
241 // the deserializer cannot get more input
242 self.failed = true;
243 return Err(Error::new(ErrorKind::EndOfFile, "unexpected end of input")
244 .shift_position(self.position));
245 }
246 return Ok(Status::NeedInput);
247 }
248 Frame::End => {
249 assert!(self.eof, "end of values before the end of the input");
250 self.done = true;
251 return Ok(Status::End);
252 }
253 }
254 }
255 }
256
257 /// Checks if another value follows.
258 ///
259 /// Returns [`Status::Ready`] if a value follows (it does not need to be
260 /// complete), [`Status::End`] if there are no more values and
261 /// [`Status::NeedInput`] if more input is needed to know. The value is
262 /// then read with [`drive_partial`](Self::drive_partial) or, once
263 /// [`poll`](Self::poll) reports it's complete, with
264 /// [`deserialize`](Self::deserialize). If the stream deserializer
265 /// cannot find the start of a value on its own (see
266 /// [`StreamDeserializer::peek`]), the value is framed which means that
267 /// it's buffered completely.
268 pub fn peek(&mut self) -> Result<Status, Error> {
269 if self.ready.is_some() || self.partial {
270 return Ok(Status::Ready);
271 }
272 if self.failed {
273 return Err(failed_error());
274 }
275 if self.done {
276 return Ok(Status::End);
277 }
278 loop {
279 let input = &self.data[self.start..self.end];
280 let progress = match self.deserializer.peek(input, self.eof) {
281 Ok(Some(progress)) => progress,
282 Ok(None) => return self.poll(),
283 Err(mut err) => {
284 self.failed = true;
285 if self.deserializer.is_text() {
286 err.resolve_position(input);
287 }
288 return Err(err.shift_position(self.position));
289 }
290 };
291 match progress {
292 Progress::Done { consumed } => {
293 assert!(consumed <= input.len(), "invalid progress");
294 self.consume(consumed);
295 return Ok(Status::Ready);
296 }
297 Progress::NeedMore { consumed } => {
298 assert!(consumed <= input.len(), "invalid progress");
299 if consumed > 0 {
300 self.consume(consumed);
301 continue;
302 }
303 if self.eof {
304 self.failed = true;
305 return Err(Error::new(ErrorKind::EndOfFile, "unexpected end of input")
306 .shift_position(self.position));
307 }
308 return Ok(Status::NeedInput);
309 }
310 Progress::End => {
311 assert!(self.eof, "end of values before the end of the input");
312 self.done = true;
313 return Ok(Status::End);
314 }
315 }
316 }
317 }
318
319 /// Returns `true` if the stream deserializer can deserialize values
320 /// while their input arrives.
321 ///
322 /// See [`StreamDeserializer::supports_partial`] and
323 /// [`drive_partial`](Self::drive_partial).
324 pub fn supports_partial(&self) -> bool {
325 self.deserializer.supports_partial()
326 }
327
328 /// Deserializes the next value in parts while its input arrives.
329 ///
330 /// This is the alternative to [`poll`](Self::poll) and
331 /// [`deserialize`](Self::deserialize) for stream deserializers which
332 /// support it (see [`supports_partial`](Self::supports_partial)) and values
333 /// which do not borrow from the input. If the value was framed already
334 /// (by [`peek`](Self::peek) of a format that cannot find the start of a
335 /// value otherwise), it's deserialized from its frame. The input is fed into the driver until the
336 /// value is complete ([`Status::Ready`]), the input is consumed as it's
337 /// used. If more input is needed ([`Status::NeedInput`]) the method has
338 /// to be invoked again with the same driver once more input was read.
339 /// In the meantime the buffer cannot be used otherwise. After an error
340 /// the value is abandoned, whether the stream can continue with the next
341 /// value depends on the stream deserializer.
342 ///
343 /// ```
344 /// # use deser::de::{Frame, Progress, StreamDeserializer};
345 /// # use deser::Error;
346 /// # /// A format with sequences of digits (without separators).
347 /// # #[derive(Default)]
348 /// # struct Digits { started: bool }
349 /// # impl StreamDeserializer for Digits {
350 /// # fn frame(&mut self, _: &[u8], _: bool) -> Result<Frame, Error> { unimplemented!() }
351 /// # fn drive_frame<'de>(&mut self, _: &'de [u8], _: &mut DeserializeDriver<'_, 'de>) -> Result<(), Error> { unimplemented!() }
352 /// # fn supports_partial(&self) -> bool { true }
353 /// # fn drive_partial(&mut self, input: &[u8], _: usize, eof: bool, driver: &mut DeserializeDriver<'_, '_>) -> Result<Progress, Error> {
354 /// # if !self.started {
355 /// # if input.is_empty() && eof { return Ok(Progress::End); }
356 /// # driver.emit(deser::Event::seq_start())?;
357 /// # self.started = true;
358 /// # }
359 /// # for digit in input { driver.emit(u64::from(digit - b'0'))?; }
360 /// # if eof {
361 /// # driver.emit(deser::Event::SeqEnd)?;
362 /// # self.started = false;
363 /// # return Ok(Progress::Done { consumed: input.len() });
364 /// # }
365 /// # Ok(Progress::NeedMore { consumed: input.len() })
366 /// # }
367 /// # }
368 /// use deser::de::DeserializeDriver;
369 /// use deser::stream::{InputBuffer, Status};
370 ///
371 /// // `Digits` is the stream deserializer of a format with a sequence
372 /// // of digits
373 /// let mut buffer = InputBuffer::new(Digits::default());
374 /// let mut out = None::<Vec<u32>>;
375 /// {
376 /// let mut driver = DeserializeDriver::new(&mut out);
377 /// for chunk in [&b"12"[..], b"3"] {
378 /// buffer.extend_from_slice(chunk);
379 /// assert_eq!(buffer.drive_partial(&mut driver).unwrap(), Status::NeedInput);
380 /// }
381 /// buffer.set_eof();
382 /// assert_eq!(buffer.drive_partial(&mut driver).unwrap(), Status::Ready);
383 /// }
384 /// assert_eq!(out.unwrap(), [1, 2, 3]);
385 /// ```
386 ///
387 /// # Panics
388 ///
389 /// Panics if the stream deserializer does not support partial
390 /// deserialization.
391 pub fn drive_partial(
392 &mut self,
393 driver: &mut DeserializeDriver<'_, '_>,
394 ) -> Result<Status, Error> {
395 assert!(
396 self.deserializer.supports_partial(),
397 "the stream deserializer does not support partial deserialization"
398 );
399 self.apply_context(driver);
400 // a value that was framed already (see `peek`)
401 if self.ready.is_some() {
402 return self.drive_transient(driver).map(|()| Status::Ready);
403 }
404 if self.failed {
405 return Err(failed_error());
406 }
407 if self.done {
408 return Ok(Status::End);
409 }
410 let input = &self.data[self.start..self.end];
411 let rv = self
412 .deserializer
413 .drive_partial(input, self.position.offset, self.eof, driver);
414 match rv {
415 Ok(Progress::Done { consumed }) => {
416 assert!(consumed <= input.len(), "invalid progress");
417 self.consume(consumed);
418 self.partial = false;
419 Ok(Status::Ready)
420 }
421 Ok(Progress::NeedMore { consumed }) => {
422 assert!(consumed <= input.len(), "invalid progress");
423 self.consume(consumed);
424 self.partial = true;
425 if self.eof {
426 self.failed = true;
427 return Err(self.locate(Error::with_offset(
428 ErrorKind::EndOfFile,
429 "unexpected end of input",
430 self.position.offset,
431 )));
432 }
433 Ok(Status::NeedInput)
434 }
435 Ok(Progress::End) => {
436 assert!(self.eof, "end of values before the end of the input");
437 self.done = true;
438 self.partial = false;
439 Ok(Status::End)
440 }
441 // whether the stream can continue is up to the deserializer
442 Err(err) => {
443 self.partial = false;
444 Err(self.locate(err))
445 }
446 }
447 }
448
449 /// Resolves the line and column of an error with an offset in the
450 /// stream.
451 ///
452 /// This is only possible for offsets in the buffered data.
453 fn locate(&self, err: Error) -> Error {
454 err.map_each(|mut err| {
455 if let Some(offset) = err.offset()
456 && self.deserializer.is_text()
457 && err.line().is_none()
458 && offset >= self.position.offset
459 && offset - self.position.offset <= self.end - self.start
460 {
461 let mut position = self.position;
462 position
463 .advance(&self.data[self.start..self.start + offset - self.position.offset]);
464 err.set_position(offset, position.line, position.column);
465 }
466 err
467 })
468 }
469
470 /// Returns the buffer to read the next input into.
471 ///
472 /// After data was placed in the buffer, [`filled`](Self::filled) has
473 /// to be called with its length. The buffer is never empty.
474 pub fn read_buf(&mut self) -> &mut [u8] {
475 if self.data.len() - self.end < READ_SIZE {
476 // move the unconsumed input to the front before growing
477 if self.start > 0 {
478 self.data.copy_within(self.start..self.end, 0);
479 self.end -= self.start;
480 self.start = 0;
481 }
482 if self.data.len() - self.end < READ_SIZE {
483 let len = (self.end + READ_SIZE).max(self.data.len() * 2);
484 // a zeroed allocation instead of resizing, which writes the
485 // zeroes one by one without optimizations (and in miri)
486 let mut data = vec![0; len];
487 data[..self.end].copy_from_slice(&self.data[..self.end]);
488 self.data = data;
489 }
490 }
491 &mut self.data[self.end..]
492 }
493
494 /// Adds data that was read into [`read_buf`](Self::read_buf).
495 ///
496 /// # Panics
497 ///
498 /// Panics if the length exceeds the buffer or if the end of the stream
499 /// was reached.
500 pub fn filled(&mut self, len: usize) {
501 assert!(!self.eof, "data after the end of the stream");
502 assert!(
503 len <= self.data.len() - self.end,
504 "more data than read into"
505 );
506 self.end += len;
507 }
508
509 /// Marks the end of the stream.
510 pub fn set_eof(&mut self) {
511 self.eof = true;
512 }
513
514 /// Adds input by copying it into the buffer.
515 ///
516 /// This is an alternative to [`read_buf`](Self::read_buf) and
517 /// [`filled`](Self::filled) for input that is already in memory.
518 pub fn extend_from_slice(&mut self, mut input: &[u8]) {
519 while !input.is_empty() {
520 let buf = self.read_buf();
521 let len = buf.len().min(input.len());
522 buf[..len].copy_from_slice(&input[..len]);
523 self.filled(len);
524 input = &input[len..];
525 }
526 }
527
528 /// Takes the frame of the ready value.
529 ///
530 /// Returns the range of the frame in the data and its position.
531 fn take_ready(&mut self) -> (core::ops::Range<usize>, Position) {
532 let (start, end, consumed) = self
533 .ready
534 .take()
535 .expect("no value is ready, poll the buffer first");
536 let mut position = self.position;
537 position.advance(&self.data[self.start..self.start + start]);
538 let range = self.start + start..self.start + end;
539 self.consume(consumed);
540 (range, position)
541 }
542
543 /// Deserializes the ready value.
544 ///
545 /// The value can borrow from the buffer.
546 ///
547 /// # Panics
548 ///
549 /// Panics if no value is ready (see [`poll`](Self::poll)).
550 pub fn deserialize<'a, T: Deserialize<'a>>(&'a mut self) -> Result<T, Error> {
551 self.deserialize_with(|_| {})
552 }
553
554 /// Deserializes the ready value with a configured driver.
555 ///
556 /// The callback is invoked with the driver before the value is
557 /// deserialized, for instance to add [`Layer`](crate::de::Layer)s.
558 ///
559 /// # Panics
560 ///
561 /// Panics if no value is ready (see [`poll`](Self::poll)).
562 pub fn deserialize_with<'a, T, F>(&'a mut self, setup: F) -> Result<T, Error>
563 where
564 T: Deserialize<'a>,
565 F: FnOnce(&mut DeserializeDriver<'_, 'a>),
566 {
567 crate::de::deserialize_value(|driver| {
568 setup(driver);
569 self.drive(driver)
570 })
571 }
572
573 /// Feeds the events of the ready value into a driver.
574 ///
575 /// This is useful to deserialize into a custom
576 /// [`Sink`](crate::de::Sink). The value can borrow from the buffer.
577 /// To feed it into a driver which outlives the buffer's data (for
578 /// instance to implement [`Deserializer`](crate::de::Deserializer) for
579 /// a reader), lend the driver out with
580 /// [`DeserializeDriver::transient`]:
581 ///
582 /// ```
583 /// # use deser::de::{DeserializeDriver, StreamDeserializer};
584 /// # use deser::stream::InputBuffer;
585 /// # use deser::Error;
586 /// fn drive<D: StreamDeserializer>(
587 /// buffer: &mut InputBuffer<D>,
588 /// driver: &mut DeserializeDriver<'_, '_>,
589 /// ) -> Result<(), Error> {
590 /// driver.transient(|driver| buffer.drive(driver))
591 /// }
592 /// ```
593 ///
594 /// # Panics
595 ///
596 /// Panics if no value is ready (see [`poll`](Self::poll)).
597 pub fn drive<'a>(&'a mut self, driver: &mut DeserializeDriver<'_, 'a>) -> Result<(), Error> {
598 self.apply_context(driver);
599 let (range, position) = self.take_ready();
600 let frame = &self.data[range];
601 self.deserializer
602 .drive_frame(frame, driver)
603 .map_err(|err| err.shift_position(position))
604 }
605
606 /// Feeds the events of the ready value into a driver for any lifetime.
607 ///
608 /// This is like [`drive`](Self::drive) but the value cannot borrow
609 /// from the buffer: the driver is lent out with
610 /// [`DeserializeDriver::transient`].
611 pub(crate) fn drive_transient(
612 &mut self,
613 driver: &mut DeserializeDriver<'_, '_>,
614 ) -> Result<(), Error> {
615 let (range, position) = self.take_ready();
616 let frame = &self.data[range];
617 let deserializer = &mut self.deserializer;
618 driver
619 .transient(|driver| deserializer.drive_frame(frame, driver))
620 .map_err(|err| err.shift_position(position))
621 }
622
623 /// Creates the error for a value where none is expected.
624 ///
625 /// The error refers to the start of the ready value. Adapters use
626 /// this to check that a stream ends after a value (see
627 /// `Reader::end` of `deser::io`).
628 ///
629 /// # Panics
630 ///
631 /// Panics if no value is ready (see [`poll`](Self::poll)).
632 pub fn trailing_error(&self) -> Error {
633 let (start, _, _) = self.ready.expect("no value is ready");
634 let mut position = self.position;
635 position.advance(&self.data[self.start..self.start + start]);
636 Error::with_position(ErrorKind::Syntax, "unexpected value after the end", 0, 1, 1)
637 .shift_position(position)
638 }
639}
640
641#[cold]
642fn failed_error() -> Error {
643 Error::new(ErrorKind::InvalidState, "cannot continue after an error")
644}