1use crate::{
4 error::EdifactError,
5 model::{Element, OwnedSegment, Segment, Span},
6 tokenizer::{Token, Tokenizer},
7};
8use memchr::memchr2;
9use smallvec::SmallVec;
10use std::borrow::Cow;
11use std::io::{BufRead, BufReader, Read};
12
13fn finish_element<'a>(
14 elements: &mut Vec<Element<'a>>,
15 current_components: &mut SmallVec<[Cow<'a, str>; 4]>,
16 current_component_spans: &mut SmallVec<[Span; 4]>,
17 current_element_start: &mut Option<usize>,
18) {
19 if let (Some(start), Some(last_span)) = (
20 current_element_start.take(),
21 current_component_spans.last().copied(),
22 ) {
23 elements.push(Element {
24 span: Span::new(start, last_span.end),
25 components: std::mem::take(current_components),
26 component_spans: std::mem::take(current_component_spans),
27 });
28 }
29}
30
31fn resolve_release(
32 val: &str,
33 release_char: char,
34 start_offset: usize,
35) -> Result<Cow<'_, str>, EdifactError> {
36 if !val.contains(release_char) {
37 return Ok(Cow::Borrowed(val));
38 }
39 resolve_release_owned(val, release_char, start_offset).map(Cow::Owned)
40}
41
42fn resolve_release_owned(
43 val: &str,
44 release_char: char,
45 start_offset: usize,
46) -> Result<String, EdifactError> {
47 let mut out = String::with_capacity(val.len());
48 let mut chars = val.chars();
49 while let Some(c) = chars.next() {
50 if c == release_char {
51 if let Some(escaped) = chars.next() {
52 out.push(escaped);
53 } else {
54 return Err(EdifactError::InvalidReleaseSequence {
55 offset: start_offset + val.len().saturating_sub(1),
56 });
57 }
58 } else {
59 out.push(c);
60 }
61 }
62 Ok(out)
63}
64
65pub struct Parser<'a> {
70 tokenizer: Tokenizer<'a>,
71 peeked: Option<Token<'a>>,
73 release_char: char,
75}
76
77impl<'a> Parser<'a> {
78 pub fn new(tokenizer: Tokenizer<'a>) -> Self {
80 let release_char = tokenizer.service_string_advice().release_char as char;
81 Self {
82 tokenizer,
83 peeked: None,
84 release_char,
85 }
86 }
87}
88
89impl<'a> Iterator for Parser<'a> {
90 type Item = Result<Segment<'a>, EdifactError>;
91
92 fn next(&mut self) -> Option<Self::Item> {
93 let (tag, tag_span) = loop {
95 let tok = match self.peeked.take() {
96 Some(t) => Ok(t),
97 None => self.tokenizer.next()?,
98 };
99 match tok {
100 Ok(Token::SegmentTag { value, span }) => break (value, span),
101 Ok(Token::SegmentTerminator { .. }) => continue, Ok(_) => continue, Err(e) => return Some(Err(e)),
104 }
105 };
106
107 let mut elements: Vec<Element<'a>> = Vec::with_capacity(8);
108 let mut current_components: SmallVec<[Cow<'a, str>; 4]> = SmallVec::new();
109 let mut current_component_spans: SmallVec<[Span; 4]> = SmallVec::new();
110 let mut current_element_start: Option<usize> = None;
111 let mut in_element = false;
112 let mut segment_end = tag_span.end;
113
114 loop {
115 let tok = match self.tokenizer.next() {
116 Some(Ok(t)) => t,
117 Some(Err(e)) => return Some(Err(e)),
118 None => {
119 if in_element {
121 finish_element(
122 &mut elements,
123 &mut current_components,
124 &mut current_component_spans,
125 &mut current_element_start,
126 );
127 if let Some(last) = elements.last() {
128 segment_end = last.span.end;
129 }
130 }
131 break;
132 }
133 };
134
135 match tok {
136 Token::SegmentTag {
137 value: next_tag,
138 span,
139 } => {
140 self.peeked = Some(Token::SegmentTag {
142 value: next_tag,
143 span,
144 });
145 if in_element {
146 finish_element(
147 &mut elements,
148 &mut current_components,
149 &mut current_component_spans,
150 &mut current_element_start,
151 );
152 if let Some(last) = elements.last() {
153 segment_end = last.span.end;
154 }
155 }
156 break;
157 }
158 Token::SegmentTerminator { span } => {
159 if in_element {
160 finish_element(
161 &mut elements,
162 &mut current_components,
163 &mut current_component_spans,
164 &mut current_element_start,
165 );
166 }
167 segment_end = span.end;
168 break;
169 }
170 Token::DataElement { value, span } => {
171 if in_element {
172 finish_element(
173 &mut elements,
174 &mut current_components,
175 &mut current_component_spans,
176 &mut current_element_start,
177 );
178 }
179 let resolved = match resolve_release(value, self.release_char, span.start) {
180 Ok(v) => v,
181 Err(error) => return Some(Err(error)),
182 };
183 current_components.push(resolved);
184 current_component_spans.push(span);
185 current_element_start = Some(span.start);
186 in_element = true;
187 }
188 Token::ComponentElement { value, span } => {
189 if !in_element {
190 in_element = true;
192 current_element_start = Some(span.start);
193 }
194 let resolved = match resolve_release(value, self.release_char, span.start) {
195 Ok(v) => v,
196 Err(error) => return Some(Err(error)),
197 };
198 current_components.push(resolved);
199 current_component_spans.push(span);
200 }
201 }
202 }
203
204 Some(Ok(Segment {
205 tag,
206 span: Span::new(tag_span.start, segment_end),
207 tag_span,
208 elements,
209 }))
210 }
211}
212
213pub fn from_reader<R: Read>(reader: R) -> Result<Vec<OwnedSegment>, EdifactError> {
219 from_reader_stream(reader).collect()
220}
221
222pub fn from_bufread<R: BufRead>(reader: R) -> Result<Vec<OwnedSegment>, EdifactError> {
224 from_bufread_stream(reader).collect()
225}
226
227#[derive(Debug, Clone, Copy)]
243pub struct ReaderConfig {
244 pub max_segment_bytes: usize,
254}
255
256impl Default for ReaderConfig {
257 fn default() -> Self {
258 Self {
259 max_segment_bytes: 65_536,
260 }
261 }
262}
263
264impl ReaderConfig {
265 #[must_use]
267 pub fn max_segment_bytes(mut self, limit: usize) -> Self {
268 self.max_segment_bytes = limit;
269 self
270 }
271}
272
273#[derive(Debug, Clone, Copy, PartialEq, Eq)]
275enum StreamState {
276 Init,
278 Running,
280 Done,
282}
283
284pub struct OwnedSegmentStream<R: BufRead> {
302 reader: R,
303 ssa: crate::tokenizer::ServiceStringAdvice,
304 state: StreamState,
305 stream_offset: usize,
306 config: ReaderConfig,
307}
308
309impl<R: BufRead> OwnedSegmentStream<R> {
310 fn new(reader: R) -> Self {
311 Self::with_config(reader, ReaderConfig::default())
312 }
313
314 fn with_config(reader: R, config: ReaderConfig) -> Self {
315 Self {
316 reader,
317 ssa: crate::tokenizer::ServiceStringAdvice::default(),
318 state: StreamState::Init,
319 stream_offset: 0,
320 config,
321 }
322 }
323}
324
325enum FastSegment {
329 Parsed(OwnedSegment, usize),
331 Skip(usize),
333 NeedMore,
335 Eof,
337 Err(EdifactError),
339}
340
341fn find_unescaped_term(buf: &[u8], term: u8, release: u8) -> Option<usize> {
353 let mut i = 0;
354 while i < buf.len() {
355 let rel = memchr2(release, term, &buf[i..])?;
357 let pos = i + rel;
358 if buf[pos] == release {
359 i = pos + 2;
361 } else {
362 return Some(pos);
364 }
365 }
366 None
367}
368
369fn try_fast_segment<R: BufRead>(
374 reader: &mut R,
375 ssa: crate::tokenizer::ServiceStringAdvice,
376 seg_start: usize,
377 max_segment_bytes: usize,
378) -> FastSegment {
379 let buf = match reader.fill_buf() {
380 Ok(b) => b,
381 Err(e) => return FastSegment::Err(e.into()),
382 };
383
384 if buf.is_empty() {
385 return FastSegment::Eof;
386 }
387
388 let Some(pos) = find_unescaped_term(buf, ssa.segment_term, ssa.release_char) else {
389 return FastSegment::NeedMore;
390 };
391
392 if pos > max_segment_bytes {
395 return FastSegment::Err(EdifactError::SegmentTooLong {
396 offset: seg_start,
397 limit: max_segment_bytes,
398 });
399 }
400
401 let seg_bytes = &buf[..pos];
403
404 if seg_bytes
406 .iter()
407 .all(|&b| matches!(b, b' ' | b'\t' | b'\r' | b'\n'))
408 {
409 return FastSegment::Skip(pos + 1);
410 }
411
412 let tok = Tokenizer::new(&buf[..pos + 1], ssa);
416 match Parser::new(tok).collect::<Result<Vec<Segment<'_>>, _>>() {
417 Err(e) => FastSegment::Err(e),
418 Ok(segs) => match segs.into_iter().next() {
419 None => FastSegment::Skip(pos + 1),
420 Some(s) => {
421 FastSegment::Parsed(OwnedSegment::from(s).offset(seg_start), pos + 1)
422 }
423 },
424 }
425 }
427
428impl<R: BufRead> Iterator for OwnedSegmentStream<R> {
431 type Item = Result<OwnedSegment, EdifactError>;
432
433 fn next(&mut self) -> Option<Self::Item> {
434 if self.state == StreamState::Done {
435 return None;
436 }
437
438 loop {
439 if self.state == StreamState::Running {
441 let seg_start = self.stream_offset;
442 match try_fast_segment(&mut self.reader, self.ssa, seg_start, self.config.max_segment_bytes) {
443 FastSegment::Parsed(seg, n) => {
444 self.reader.consume(n);
445 self.stream_offset += n;
446 return Some(Ok(seg));
447 }
448 FastSegment::Skip(n) => {
449 self.reader.consume(n);
450 self.stream_offset += n;
451 continue;
452 }
453 FastSegment::Eof => return None,
454 FastSegment::Err(e) => {
455 self.state = StreamState::Done;
456 return Some(Err(e));
457 }
458 FastSegment::NeedMore => {
459 }
461 }
462 }
463
464 let mut scanned = self.state != StreamState::Init;
466 let mut raw = match read_next_raw_segment(
467 &mut self.reader,
468 &mut self.ssa,
469 &mut scanned,
470 &mut self.stream_offset,
471 self.config.max_segment_bytes,
472 ) {
473 Ok(Some(r)) => r,
474 Ok(None) => return None,
475 Err(e) => {
476 self.state = StreamState::Done;
477 return Some(Err(e));
478 }
479 };
480 if scanned {
481 self.state = StreamState::Running;
482 }
483
484 raw.bytes.push(self.ssa.segment_term);
485 let tok = Tokenizer::new(raw.bytes.as_slice(), self.ssa);
486 match Parser::new(tok).collect::<Result<Vec<Segment<'_>>, _>>() {
487 Ok(segs) => {
488 if let Some(s) = segs.into_iter().next() {
489 return Some(Ok(OwnedSegment::from(s).offset(raw.start_offset)));
490 }
491 }
493 Err(e) => {
494 self.state = StreamState::Done;
495 return Some(Err(e));
496 }
497 }
498 }
499 }
500}
501
502pub fn from_bufread_stream<R: BufRead>(reader: R) -> OwnedSegmentStream<R> {
504 OwnedSegmentStream::new(reader)
505}
506
507pub fn from_bufread_stream_with_config<R: BufRead>(
509 reader: R,
510 config: ReaderConfig,
511) -> OwnedSegmentStream<R> {
512 OwnedSegmentStream::with_config(reader, config)
513}
514
515pub fn from_reader_stream<R: Read>(reader: R) -> OwnedSegmentStream<BufReader<R>> {
517 from_bufread_stream(BufReader::new(reader))
518}
519
520pub fn from_reader_with_config<R: Read>(
533 reader: R,
534 config: ReaderConfig,
535) -> OwnedSegmentStream<BufReader<R>> {
536 from_bufread_stream_with_config(BufReader::new(reader), config)
537}
538
539fn read_next_raw_segment<R: BufRead>(
540 reader: &mut R,
541 ssa: &mut crate::tokenizer::ServiceStringAdvice,
542 scanned_header: &mut bool,
543 stream_offset: &mut usize,
544 max_segment_bytes: usize,
545) -> Result<Option<crate::tokenizer::RawSegment>, EdifactError> {
546 loop {
547 let Some((first_offset, first)) = read_next_non_ws_byte(reader, stream_offset)? else {
548 return Ok(None);
549 };
550
551 if !*scanned_header && first == b'U' {
552 let second = read_required_byte(reader, stream_offset)?;
553 let third = read_required_byte(reader, stream_offset)?;
554 if second == b'N' && third == b'A' {
555 let mut una = [0u8; 9];
556 una[0] = b'U';
557 una[1] = b'N';
558 una[2] = b'A';
559 for slot in una.iter_mut().skip(3) {
560 *slot = read_required_byte(reader, stream_offset)?;
561 }
562 *ssa = crate::tokenizer::ServiceStringAdvice {
563 component_sep: una[3],
564 element_sep: una[4],
565 decimal_mark: una[5],
566 release_char: una[6],
567 segment_term: una[8],
568 };
569 if !ssa.is_valid() {
570 return Err(EdifactError::InvalidUna);
571 }
572 *scanned_header = true;
573 continue;
574 }
575
576 *scanned_header = true;
577 return read_remainder_of_segment(
578 reader,
579 ssa,
580 crate::tokenizer::RawSegment {
581 bytes: vec![first, second, third],
582 start_offset: first_offset,
583 },
584 stream_offset,
585 max_segment_bytes,
586 );
587 }
588
589 *scanned_header = true;
590 return read_remainder_of_segment(
591 reader,
592 ssa,
593 crate::tokenizer::RawSegment {
594 bytes: vec![first],
595 start_offset: first_offset,
596 },
597 stream_offset,
598 max_segment_bytes,
599 );
600 }
601}
602
603fn read_remainder_of_segment<R: BufRead>(
604 reader: &mut R,
605 ssa: &crate::tokenizer::ServiceStringAdvice,
606 mut out: crate::tokenizer::RawSegment,
607 stream_offset: &mut usize,
608 max_segment_bytes: usize,
609) -> Result<Option<crate::tokenizer::RawSegment>, EdifactError> {
610 let mut escaped = false;
611 loop {
612 if out.bytes.len() >= max_segment_bytes {
613 return Err(EdifactError::SegmentTooLong {
614 offset: out.start_offset,
615 limit: max_segment_bytes,
616 });
617 }
618 let Some(byte) = read_next_byte(reader, stream_offset)? else {
619 return if out.bytes.is_empty() {
620 Ok(None)
621 } else if escaped {
622 Err(EdifactError::InvalidReleaseSequence {
623 offset: out.start_offset + out.bytes.len().saturating_sub(1),
624 })
625 } else {
626 Err(EdifactError::UnexpectedEof {
627 offset: out.start_offset + out.bytes.len(),
628 })
629 };
630 };
631
632 if !escaped && byte == ssa.segment_term {
633 return Ok(Some(out));
634 }
635
636 if !escaped && byte == ssa.release_char {
637 escaped = true;
638 out.bytes.push(byte);
639 continue;
640 }
641
642 escaped = false;
643 out.bytes.push(byte);
644 }
645}
646
647fn read_next_byte<R: BufRead>(
648 reader: &mut R,
649 stream_offset: &mut usize,
650) -> Result<Option<u8>, EdifactError> {
651 let buf = reader.fill_buf()?;
652 if buf.is_empty() {
653 return Ok(None);
654 }
655
656 let byte = buf[0];
657 reader.consume(1);
658 *stream_offset += 1;
659 Ok(Some(byte))
660}
661
662fn read_required_byte<R: BufRead>(
663 reader: &mut R,
664 stream_offset: &mut usize,
665) -> Result<u8, EdifactError> {
666 read_next_byte(reader, stream_offset)?.ok_or(EdifactError::UnexpectedEof {
667 offset: *stream_offset,
668 })
669}
670
671fn read_next_non_ws_byte<R: BufRead>(
672 reader: &mut R,
673 stream_offset: &mut usize,
674) -> Result<Option<(usize, u8)>, EdifactError> {
675 loop {
676 let current_offset = *stream_offset;
677 let Some(byte) = read_next_byte(reader, stream_offset)? else {
678 return Ok(None);
679 };
680 if !matches!(byte, b' ' | b'\t' | b'\r' | b'\n') {
681 return Ok(Some((current_offset, byte)));
682 }
683 }
684}
685
686#[cfg(test)]
687mod tests {
688 use super::*;
689 use crate::tokenizer::ServiceStringAdvice;
690
691 fn parse_all(input: &[u8]) -> Vec<Segment<'_>> {
692 let ssa = ServiceStringAdvice::from_bytes(input);
693 let tok = Tokenizer::new(input, ssa);
694 Parser::new(tok)
695 .collect::<Result<Vec<_>, _>>()
696 .expect("parse failed")
697 }
698
699 #[test]
700 fn parses_unb_unz() {
701 let input = b"UNB+UNOA:1+SENDER+RECEIVER+200101:0900+1'UNZ+0+1'";
702 let segs = parse_all(input);
703 assert_eq!(segs.len(), 2);
704 assert_eq!(segs[0].tag, "UNB");
705 assert_eq!(segs[1].tag, "UNZ");
706 assert_eq!(segs[0].tag_span, Span::new(0, 3));
707 assert_eq!(segs[0].span, Span::new(0, 41));
708 }
709
710 #[test]
711 fn element_access() {
712 let input = b"BGM+220+ORDER123+9'";
713 let segs = parse_all(input);
714 assert_eq!(segs[0].element_str(0), Some("220"));
715 assert_eq!(segs[0].element_str(1), Some("ORDER123"));
716 }
717
718 #[test]
719 fn component_access() {
720 let input = b"DTM+137:20200101:102'";
721 let segs = parse_all(input);
722 let dtm = &segs[0];
723 assert_eq!(dtm.get_element(0).unwrap().get_component(0), Some("137"));
724 assert_eq!(
725 dtm.get_element(0).unwrap().get_component(1),
726 Some("20200101")
727 );
728 assert_eq!(dtm.get_element(0).unwrap().get_component(2), Some("102"));
729 }
730
731 #[test]
732 fn release_char_resolved() {
733 let input = b"FTX+AAA++test?+value'";
734 let segs = parse_all(input);
735 assert_eq!(segs[0].element_str(2), Some("test+value"));
736 assert_eq!(
737 segs[0].get_element(2).unwrap().component_span(0),
738 Some(Span::new(9, 20))
739 );
740 }
741
742 #[test]
743 fn reader_path_preserves_custom_una_delimiters() {
744 let input = b"UNA:;.? 'BGM;220;test?;value'";
745 let segments = super::from_bufread(std::io::BufReader::new(std::io::Cursor::new(input)))
746 .expect("reader parse should succeed");
747 let bgm = segments
748 .iter()
749 .find(|segment| segment.tag == "BGM")
750 .expect("BGM segment should be present");
751 assert_eq!(bgm.elements[0].components[0], "220");
752 assert_eq!(bgm.elements[1].components[0], "test;value");
753 }
754
755 #[test]
756 fn arbitrary_bytes_no_panic() {
757 let garbage: &[u8] = b"\xff\x00\x01\x02ABC+++'''???";
759 let _ = crate::from_bytes(garbage).collect::<Vec<_>>();
760 }
761
762 #[test]
763 fn from_reader_handles_chunk_boundaries() {
764 let input = b"UNA:+.? 'BGM+220+test?+value'UNT+2+1'";
765 let reader = std::io::BufReader::with_capacity(5, std::io::Cursor::new(input));
766 let parsed = from_bufread(reader).expect("reader parsing should succeed");
767 assert_eq!(parsed.len(), 2);
768 assert_eq!(parsed[0].tag, "BGM");
769 assert_eq!(parsed[0].elements[1].components[0], "test+value");
770 assert_eq!(parsed[1].tag, "UNT");
771 }
772
773 #[test]
774 fn from_reader_without_una_uses_default_delimiters() {
775 let input = b"BGM+220+X'UNT+2+1'";
776 let parsed =
777 from_reader(std::io::Cursor::new(input)).expect("reader parsing should succeed");
778 assert_eq!(parsed.len(), 2);
779 assert_eq!(parsed[0].tag, "BGM");
780 assert_eq!(parsed[0].elements[0].components[0], "220");
781 assert_eq!(parsed[1].span, Span::new(10, 18));
782 }
783
784 #[test]
785 fn dangling_release_sequence_is_error() {
786 let input = b"FTX+AAA++dangling?";
787 let err = crate::from_bytes(input)
788 .collect::<Result<Vec<_>, _>>()
789 .expect_err("expected dangling release to fail");
790
791 assert!(matches!(err, EdifactError::InvalidReleaseSequence { .. }));
792 }
793
794 #[test]
795 fn from_reader_reports_dangling_release_sequence() {
796 let input = b"FTX+AAA++dangling?";
797 let err = from_reader(std::io::Cursor::new(input))
798 .expect_err("expected dangling release from reader path");
799 assert!(matches!(err, EdifactError::InvalidReleaseSequence { .. }));
800 }
801
802 #[test]
803 fn from_reader_rejects_invalid_una() {
804 let input = b"UNA::.? 'BGM:220'";
805 let err = from_reader(std::io::Cursor::new(input))
806 .expect_err("invalid UNA should fail reader parsing");
807 assert!(matches!(err, EdifactError::InvalidUna));
808 }
809}