1#[cfg(feature = "io")]
5use std::io::Read;
6
7#[cfg(feature = "io")]
8use deser_core::de::DeserializeOwned;
9use deser_core::de::{self, DeserializeDriver, Frame, Progress};
10use deser_core::{Error, ErrorKind, State};
11
12use crate::Trailing;
13use crate::de::{Deserializer, DeserializerConfig};
14use crate::parser::Cursor;
15use crate::parser::{Copying, Discard, Options, Parser, Progress as ParseProgress};
16use crate::scan::LineScan;
17use crate::scan::skip_to_escape;
18
19fn skip_whitespace(input: &[u8], pos: usize, eof: bool) -> (usize, bool) {
25 let mut cursor = Cursor::new_partial(input, pos, eof);
26 let token = cursor.parse_whitespace().is_some();
27 (cursor.pos, token)
28}
29
30#[derive(Debug, Default)]
32struct StreamState {
33 done: bool,
35 parser: Parser,
37 skipping: Option<usize>,
40 failed: bool,
42 ended: bool,
44 pos: usize,
46 line: LineScan,
48 value: Option<Value>,
50}
51
52#[derive(Debug)]
54struct Value {
55 start: usize,
56 kind: ValueKind,
57 depth: usize,
58 in_string: bool,
59}
60
61#[derive(Debug)]
62enum ValueKind {
63 Scalar { parser: Parser, parsed: usize },
66 Structure,
68}
69
70fn frame_all(state: &mut StreamState, input: &[u8], eof: bool) -> Result<Frame, Error> {
71 if state.done {
72 return trailing_whitespace(input, 0, eof).map(|progress| match progress {
74 Progress::End => Frame::End,
75 Progress::NeedMore { consumed } => Frame::Incomplete { consumed },
77 Progress::Done { .. } => unreachable!(),
78 });
79 }
80 if !eof {
81 return Ok(Frame::Incomplete { consumed: 0 });
82 }
83 let (start, _) = skip_whitespace(input, 0, eof);
85 Ok(if start < input.len() {
86 state.done = true;
87 Frame::Value {
88 start,
89 end: input.len(),
90 consumed: input.len(),
91 }
92 } else {
93 Frame::End
94 })
95}
96
97fn trailing_whitespace(input: &[u8], offset: usize, eof: bool) -> Result<Progress, Error> {
101 match skip_whitespace(input, 0, eof) {
102 (pos, _) if pos == input.len() && eof => Ok(Progress::End),
103 (pos, false) if !eof => Ok(Progress::NeedMore { consumed: pos }),
105 (pos, _) => Err(Error::with_offset(
106 ErrorKind::Syntax,
107 "garbage after input",
108 offset + pos,
109 )),
110 }
111}
112
113fn frame_line(state: &mut StreamState, input: &[u8], eof: bool) -> Frame {
114 if state.pos == 0 {
117 match skip_whitespace(input, 0, eof) {
118 (0, true) => {}
119 (start, true) => return Frame::Incomplete { consumed: start },
120 (_, false) if input.is_empty() && eof => return Frame::End,
121 (start, false) if eof && start < input.len() => {
123 return Frame::Value {
124 start,
125 end: input.len(),
126 consumed: input.len(),
127 };
128 }
129 (start, false) => return Frame::Incomplete { consumed: start },
131 }
132 }
133 let end = state.line.find_end(input, state.pos);
135 let end = match end {
136 Some(end) => end,
137 None if eof => input.len(),
138 None => {
139 state.pos = input.len();
140 return Frame::Incomplete { consumed: 0 };
141 }
142 };
143 state.pos = 0;
144 state.line = LineScan::default();
145 let consumed = (end + 1).min(input.len());
146 match skip_whitespace(&input[..end], 0, true) {
147 (start, _) if start < end => Frame::Value {
148 start,
149 end,
150 consumed,
151 },
152 _ if consumed == 0 => Frame::End,
153 _ => Frame::Incomplete { consumed },
155 }
156}
157
158fn frame_value(state: &mut StreamState, input: &[u8], eof: bool) -> Frame {
159 let value = match state.value {
160 Some(ref mut value) => value,
161 None => {
162 let start = match skip_whitespace(input, 0, eof) {
163 (start, true) => start,
164 (_, false) if input.is_empty() && eof => return Frame::End,
165 (consumed, false) => return Frame::Incomplete { consumed },
166 };
167 let (kind, depth, in_string) = match input[start] {
168 b'"' => (ValueKind::Structure, 0, true),
169 b'{' | b'[' => (ValueKind::Structure, 1, false),
170 b'}' | b']' | b',' | b':' => {
173 return Frame::Value {
174 start,
175 end: start + 1,
176 consumed: start + 1,
177 };
178 }
179 _ => (
180 ValueKind::Scalar {
181 parser: Parser::default(),
182 parsed: 0,
183 },
184 0,
185 false,
186 ),
187 };
188 state.pos = start + 1;
189 state.value.insert(Value {
190 start,
191 kind,
192 depth,
193 in_string,
194 })
195 }
196 };
197
198 let end = match value.kind {
199 ValueKind::Scalar {
200 ref mut parser,
201 ref mut parsed,
202 } => {
203 let pos = value.start + *parsed;
204 let options = Options {
205 validate_utf8: true,
206 exact_numbers: false,
207 };
208 let mut discard = Discard(State::new());
209 match parser.parse(&input[pos..], 0, eof, 0, options, &mut discard) {
210 Ok(ParseProgress::Done(end)) => Some(pos + end),
211 Ok(ParseProgress::NeedMore(consumed)) => {
212 *parsed += consumed;
213 state.pos = input.len();
214 None
215 }
216 Err(err) => Some(error_end(&err, input, pos)),
219 }
220 }
221 ValueKind::Structure => scan_structure(input, &mut state.pos, value),
222 };
223
224 let start = value.start;
225 match end {
226 Some(end) => {
227 state.value = None;
228 state.pos = 0;
229 Frame::Value {
230 start,
231 end,
232 consumed: end,
233 }
234 }
235 None if eof => {
237 state.value = None;
238 state.pos = 0;
239 Frame::Value {
240 start,
241 end: input.len(),
242 consumed: input.len(),
243 }
244 }
245 None => {
246 value.start = 0;
248 state.pos -= start;
249 Frame::Incomplete { consumed: start }
250 }
251 }
252}
253
254fn error_end(err: &Error, input: &[u8], pos: usize) -> usize {
262 match err.offset() {
263 Some(offset) if err.kind() != ErrorKind::EndOfFile => (pos + offset + 1).min(input.len()),
264 _ => input.len(),
265 }
266}
267
268fn scan_structure(input: &[u8], pos: &mut usize, value: &mut Value) -> Option<usize> {
272 let mut index = *pos;
273 while index < input.len() {
274 if value.in_string {
275 index = skip_to_escape(input, index);
276 let byte = input.get(index).copied();
277 match byte {
278 Some(b'"') => {
279 value.in_string = false;
280 index += 1;
281 if value.depth == 0 {
282 return Some(index);
283 }
284 }
285 Some(b'\\') => {
286 if index + 1 >= input.len() {
288 break;
289 }
290 index += 2;
291 }
292 Some(_) => index += 1,
294 None => break,
295 }
296 } else {
297 match input[index] {
298 b'"' => {
299 value.in_string = true;
300 }
301 b'/' => {
304 match skip_whitespace(input, index, false) {
305 (next, true) if next == index => index += 1,
307 (next, token) => {
308 index = next;
309 if !token && next < input.len() {
310 break;
311 }
312 }
313 }
314 continue;
315 }
316 b'{' | b'[' => value.depth += 1,
317 b'}' | b']' => {
318 value.depth -= 1;
319 if value.depth == 0 {
320 return Some(index + 1);
321 }
322 }
323 _ => {}
324 }
325 index += 1;
326 }
327 }
328 *pos = index;
329 None
330}
331
332#[derive(Debug)]
373pub struct StreamDeserializer {
374 config: DeserializerConfig,
375 state: StreamState,
376}
377
378impl Default for StreamDeserializer {
379 fn default() -> StreamDeserializer {
380 StreamDeserializer::new()
381 }
382}
383
384impl StreamDeserializer {
385 pub fn new() -> StreamDeserializer {
387 StreamDeserializer::with_config(DeserializerConfig::new())
388 }
389
390 pub fn with_config(config: DeserializerConfig) -> StreamDeserializer {
392 StreamDeserializer {
393 config,
394 state: StreamState::default(),
395 }
396 }
397
398 pub fn config(&self) -> &DeserializerConfig {
400 &self.config
401 }
402
403 fn skip_to_value(
409 &mut self,
410 input: &[u8],
411 offset: usize,
412 eof: bool,
413 ) -> Result<Result<usize, Progress>, Error> {
414 let options = self.options();
415 let state = &mut self.state;
416 if state.ended {
417 return Ok(Err(Progress::End));
418 }
419 if state.failed {
420 return Err(Error::new(
421 ErrorKind::InvalidState,
422 "cannot continue after an error",
423 ));
424 }
425
426 let mut pos = 0;
428 if let Some(skip) = state.skipping {
429 let mut discard = Discard(State::new());
430 match state
431 .parser
432 .parse(input, skip, eof, offset, options, &mut discard)
433 {
434 Ok(ParseProgress::Done(end)) => {
435 state.skipping = None;
436 state.done = self.config.trailing_mode() == Trailing::Strict;
437 pos = end;
438 }
439 Ok(ParseProgress::NeedMore(consumed)) => {
440 state.skipping = Some(0);
441 return Ok(Err(Progress::NeedMore { consumed }));
442 }
443 Err(err) => {
444 state.parser.reset();
445 state.skipping = None;
446 state.failed = true;
447 return Err(err);
448 }
449 }
450 }
451
452 if !state.parser.is_idle() {
453 return Ok(Ok(pos));
454 }
455 if state.done {
456 return match trailing_whitespace(&input[pos..], offset + pos, eof)? {
457 Progress::NeedMore { consumed } => Ok(Err(Progress::NeedMore {
458 consumed: pos + consumed,
459 })),
460 progress => Ok(Err(progress)),
461 };
462 }
463 let token;
466 (pos, token) = skip_whitespace(input, pos, eof);
467 if !token && (pos == input.len() || !eof) {
468 return Ok(Err(if eof {
469 Progress::End
470 } else {
471 Progress::NeedMore { consumed: pos }
472 }));
473 }
474 Ok(Ok(pos))
475 }
476
477 fn options(&self) -> Options {
479 Options {
480 validate_utf8: true,
481 exact_numbers: self.config.exact_numbers_enabled(),
482 }
483 }
484}
485
486impl de::StreamDeserializer for StreamDeserializer {
487 fn context(&self) -> deser_core::Context {
488 self.config.context().clone()
489 }
490
491 fn frame(&mut self, input: &[u8], eof: bool) -> Result<Frame, Error> {
492 let state = &mut self.state;
493 match self.config.trailing_mode() {
494 Trailing::Strict => frame_all(state, input, eof),
495 Trailing::Newline => Ok(frame_line(state, input, eof)),
496 Trailing::Stop => Ok(frame_value(state, input, eof)),
497 }
498 }
499
500 fn drive_frame<'de>(
501 &mut self,
502 frame: &'de [u8],
503 driver: &mut DeserializeDriver<'_, 'de>,
504 ) -> Result<(), Error> {
505 let mut de = Deserializer::from_frame(frame, &self.config);
506 de.drive(driver)
507 }
508
509 fn is_text(&self) -> bool {
510 true
511 }
512
513 fn supports_partial(&self) -> bool {
516 self.config.trailing_mode() != Trailing::Newline
517 }
518
519 fn drive_partial(
520 &mut self,
521 input: &[u8],
522 offset: usize,
523 eof: bool,
524 driver: &mut DeserializeDriver<'_, '_>,
525 ) -> Result<Progress, Error> {
526 let pos = match self.skip_to_value(input, offset, eof)? {
527 Ok(pos) => pos,
528 Err(progress) => return Ok(progress),
529 };
530 let options = self.options();
531 let state = &mut self.state;
532 match state
533 .parser
534 .parse(input, pos, eof, offset, options, &mut Copying(driver))
535 {
536 Ok(ParseProgress::Done(end)) => {
537 state.done = self.config.trailing_mode() == Trailing::Strict;
538 Ok(Progress::Done { consumed: end })
539 }
540 Ok(ParseProgress::NeedMore(consumed)) => Ok(Progress::NeedMore { consumed }),
541 Err(err) => {
542 if let Some(resume) = state.parser.recoverable() {
543 state.skipping = Some(resume);
546 } else {
547 state.parser.reset();
548 state.ended = eof && err.kind() == ErrorKind::EndOfFile;
551 state.failed = true;
552 }
553 Err(err)
554 }
555 }
556 }
557
558 fn peek(&mut self, input: &[u8], eof: bool) -> Result<Option<Progress>, Error> {
559 if !de::StreamDeserializer::supports_partial(self) {
560 return Ok(None);
561 }
562 Ok(Some(match self.skip_to_value(input, 0, eof)? {
563 Ok(pos) => Progress::Done { consumed: pos },
564 Err(progress) => progress,
565 }))
566 }
567}
568
569#[cfg(feature = "io")]
570impl DeserializerConfig {
571 pub fn reader<R: Read>(&self, reader: R) -> deser_core::io::Reader<R, StreamDeserializer> {
576 deser_core::io::Reader::new(reader, StreamDeserializer::with_config(self.clone()))
577 }
578
579 pub fn from_reader<T: DeserializeOwned, R: Read>(&self, reader: R) -> Result<T, Error> {
583 deser_core::io::from_reader(reader, StreamDeserializer::with_config(self.clone()))
584 }
585}
586
587#[cfg(feature = "io")]
599pub fn from_reader<T: DeserializeOwned, R: Read>(reader: R) -> Result<T, Error> {
600 DeserializerConfig::new().from_reader(reader)
601}