freeswitch_sofia_trace_parser/
frame.rs

1use std::fmt;
2use std::io::Read;
3
4use memchr::memmem;
5use tracing::{debug, info, trace, warn};
6
7use crate::finders::BOUNDARY;
8use crate::types::{
9    Direction, Frame, ParseStats, SkipReason, SkipTracking, Timestamp, Transport, UnparsedRegion,
10};
11
12const RECV_PREFIX: &[u8] = b"recv ";
13const SENT_PREFIX: &[u8] = b"sent ";
14/// Maximum skip size classified as a partial first frame.
15/// Based on IP max datagram size (65535) plus the `\x0B\n` boundary (2 bytes).
16const MAX_PARTIAL_FRAME: usize = 65537;
17
18/// Errors produced during frame parsing (Level 1) and SIP parsing (Level 3).
19///
20/// Returned as `Iterator::Item = Result<T, ParseError>`. The caller decides
21/// whether to skip, log, or fail on each error.
22#[derive(Debug)]
23pub enum ParseError {
24    /// Frame header is malformed (e.g., missing colon, bad timestamp).
25    InvalidHeader(String),
26    /// Reassembled content is not a valid SIP message.
27    InvalidMessage(String),
28    /// Whitespace-only content from TLS/TCP keep-alive probes (RFC 5626).
29    /// Not a parse failure; can be safely ignored.
30    TransportNoise {
31        /// Number of whitespace bytes.
32        bytes: usize,
33        /// Transport of the connection that produced the noise.
34        transport: Transport,
35        /// Remote address of the connection.
36        address: String,
37    },
38    /// Underlying reader returned an I/O error.
39    Io(std::io::Error),
40}
41
42impl fmt::Display for ParseError {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        match self {
45            ParseError::InvalidHeader(msg) => write!(f, "invalid frame header: {msg}"),
46            ParseError::InvalidMessage(msg) => write!(f, "invalid SIP message: {msg}"),
47            ParseError::TransportNoise {
48                bytes,
49                transport,
50                address,
51            } => write!(
52                f,
53                "transport noise: {bytes} bytes of non-SIP data from {transport}/{address}"
54            ),
55            ParseError::Io(e) => write!(f, "I/O error: {e}"),
56        }
57    }
58}
59
60impl std::error::Error for ParseError {
61    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
62        match self {
63            ParseError::Io(e) => Some(e),
64            _ => None,
65        }
66    }
67}
68
69impl From<std::io::Error> for ParseError {
70    fn from(e: std::io::Error) -> Self {
71        ParseError::Io(e)
72    }
73}
74
75fn digit(b: u8) -> Option<u8> {
76    match b {
77        b'0'..=b'9' => Some(b - b'0'),
78        _ => None,
79    }
80}
81
82/// Parse up to `MAX_DIGITS` ASCII digits; a value the target type cannot hold
83/// is `None`, as is any non-digit byte.
84fn parse_digits<const MAX_DIGITS: usize, T: TryFrom<u64>>(bytes: &[u8]) -> Option<T> {
85    if bytes.is_empty() || bytes.len() > MAX_DIGITS {
86        return None;
87    }
88    let mut val: u64 = 0;
89    for &b in bytes {
90        val = val.checked_mul(10)?.checked_add(u64::from(digit(b)?))?;
91    }
92    T::try_from(val).ok()
93}
94
95/// Parse timestamp from bytes: either `HH:MM:SS.usec` or `YYYY-MM-DD HH:MM:SS.usec`
96fn parse_timestamp(bytes: &[u8]) -> Option<Timestamp> {
97    // Try full datetime first: YYYY-MM-DD HH:MM:SS.usec (min 26 bytes)
98    if bytes.len() >= 26 && bytes[4] == b'-' && bytes[7] == b'-' && bytes[10] == b' ' {
99        let year = parse_digits::<5, u16>(&bytes[0..4])?;
100        let month = parse_digits::<3, u8>(&bytes[5..7])?;
101        let day = parse_digits::<3, u8>(&bytes[8..10])?;
102        let ts = parse_time_part(&bytes[11..])?;
103        return Some(Timestamp::DateTime {
104            year,
105            month,
106            day,
107            hour: ts.0,
108            min: ts.1,
109            sec: ts.2,
110            usec: ts.3,
111        });
112    }
113    // Time-only: HH:MM:SS.usec (min 15 bytes)
114    let (hour, min, sec, usec) = parse_time_part(bytes)?;
115    Some(Timestamp::TimeOnly {
116        hour,
117        min,
118        sec,
119        usec,
120    })
121}
122
123/// Parse `HH:MM:SS.usec` from bytes, returns (hour, min, sec, usec)
124fn parse_time_part(bytes: &[u8]) -> Option<(u8, u8, u8, u32)> {
125    if bytes.len() < 15 {
126        return None;
127    }
128    if bytes[2] != b':' || bytes[5] != b':' || bytes[8] != b'.' {
129        return None;
130    }
131    let hour = parse_digits::<3, u8>(&bytes[0..2])?;
132    let min = parse_digits::<3, u8>(&bytes[3..5])?;
133    let sec = parse_digits::<3, u8>(&bytes[6..8])?;
134    let usec = parse_digits::<10, u32>(&bytes[9..15])?;
135    Some((hour, min, sec, usec))
136}
137
138/// Fields of one frame header line.
139#[derive(Debug)]
140pub struct FrameHeader {
141    /// Whether FreeSWITCH received or sent the frame.
142    pub direction: Direction,
143    /// Content length declared by the header.
144    pub byte_count: usize,
145    /// Transport the frame travelled over.
146    pub transport: Transport,
147    /// Remote address, as written in the header.
148    pub address: String,
149    /// Time the frame was written.
150    pub timestamp: Timestamp,
151    /// Header length in bytes, including the trailing `\n`.
152    pub header_len: usize,
153}
154
155/// Parse a frame header line from `&[u8]`.
156///
157/// Expected format:
158/// `(recv|sent) <N> bytes (from|to) <transport>/<address> at <timestamp>:\n`
159pub fn parse_frame_header(data: &[u8]) -> Result<FrameHeader, ParseError> {
160    let newline_pos = memchr::memchr(b'\n', data)
161        .ok_or_else(|| ParseError::InvalidHeader("no newline in header".into()))?;
162    let line = &data[..newline_pos];
163    // Strip trailing \r if present
164    let line = line.strip_suffix(b"\r").unwrap_or(line);
165    // Must end with ':'
166    let line = line
167        .strip_suffix(b":")
168        .ok_or_else(|| ParseError::InvalidHeader("header does not end with ':'".into()))?;
169
170    // Direction — both "recv " and "sent " are 5 bytes
171    let direction = if line.starts_with(RECV_PREFIX) {
172        Direction::Recv
173    } else if line.starts_with(SENT_PREFIX) {
174        Direction::Sent
175    } else {
176        return Err(ParseError::InvalidHeader(
177            "expected 'recv' or 'sent'".into(),
178        ));
179    };
180    let mut pos = 5;
181
182    // Byte count: digits until ' '
183    let space = memchr::memchr(b' ', &line[pos..])
184        .ok_or_else(|| ParseError::InvalidHeader("no space after byte count".into()))?;
185    let byte_count = parse_digits::<10, usize>(&line[pos..pos + space])
186        .ok_or_else(|| ParseError::InvalidHeader("invalid byte count".into()))?;
187    pos += space + 1;
188
189    // " bytes from/to "
190    let expected_recv = b"bytes from ";
191    let expected_sent = b"bytes to ";
192    if direction == Direction::Recv {
193        if !line[pos..].starts_with(expected_recv) {
194            return Err(ParseError::InvalidHeader("expected 'bytes from '".into()));
195        }
196        pos += expected_recv.len();
197    } else {
198        if !line[pos..].starts_with(expected_sent) {
199            return Err(ParseError::InvalidHeader("expected 'bytes to '".into()));
200        }
201        pos += expected_sent.len();
202    }
203
204    // Transport: tcp/ udp/ tls/ wss/
205    let transport = if line[pos..].starts_with(b"tcp/") {
206        pos += 4;
207        Transport::Tcp
208    } else if line[pos..].starts_with(b"udp/") {
209        pos += 4;
210        Transport::Udp
211    } else if line[pos..].starts_with(b"tls/") {
212        pos += 4;
213        Transport::Tls
214    } else if line[pos..].starts_with(b"wss/") {
215        pos += 4;
216        Transport::Wss
217    } else {
218        return Err(ParseError::InvalidHeader("unknown transport".into()));
219    };
220
221    // Address: until " at "
222    let at_marker = b" at ";
223    let at_pos = memmem::find(&line[pos..], at_marker)
224        .ok_or_else(|| ParseError::InvalidHeader("no ' at ' in header".into()))?;
225    let address = String::from_utf8_lossy(&line[pos..pos + at_pos]).into_owned();
226    pos += at_pos + at_marker.len();
227
228    // Timestamp: rest of line (after stripping trailing ':' already done)
229    let timestamp = parse_timestamp(&line[pos..])
230        .ok_or_else(|| ParseError::InvalidHeader("invalid timestamp".into()))?;
231
232    Ok(FrameHeader {
233        direction,
234        byte_count,
235        transport,
236        address,
237        timestamp,
238        header_len: newline_pos + 1,
239    })
240}
241
242enum HeaderParse {
243    Ok(FrameHeader),
244    NeedMore,
245    Invalid(ParseError),
246}
247
248enum SyncStep {
249    Ready,
250    End,
251    Failed(ParseError),
252}
253
254enum HeaderStep {
255    Got(FrameHeader),
256    Restart,
257    End,
258    Failed(ParseError),
259}
260
261/// Distinguish a header still arriving from one the parser rejects.
262fn classify_header(data: &[u8]) -> HeaderParse {
263    match parse_frame_header(data) {
264        Ok(header) => HeaderParse::Ok(header),
265        Err(e) => {
266            if memchr::memchr(b'\n', data).is_none() {
267                HeaderParse::NeedMore
268            } else {
269                HeaderParse::Invalid(e)
270            }
271        }
272    }
273}
274
275/// Check if data at given position looks like a valid frame header start.
276/// Used to validate `\x0B\n` boundaries.
277pub fn is_frame_header(data: &[u8]) -> bool {
278    if data.len() < 20 {
279        return false;
280    }
281    let starts_valid = data.starts_with(RECV_PREFIX) || data.starts_with(SENT_PREFIX);
282    if !starts_valid {
283        return false;
284    }
285    // Check that after direction there are digits followed by " bytes "
286    let rest = &data[5..];
287    let space = match memchr::memchr(b' ', rest) {
288        Some(p) => p,
289        None => return false,
290    };
291    if space == 0 || space > 10 {
292        return false;
293    }
294    for &b in &rest[..space] {
295        if !b.is_ascii_digit() {
296            return false;
297        }
298    }
299    rest[space..].starts_with(b" bytes ")
300}
301
302const READ_BUF_SIZE: usize = 32 * 1024;
303
304/// Level 1 streaming parser: splits raw dump bytes into [`Frame`]s.
305///
306/// Reads from any [`Read`] source and yields frames delimited by `\x0B\n`
307/// boundaries. Handles truncated first/last frames, file concatenation,
308/// and garbage recovery.
309///
310/// # Example
311///
312/// ```no_run
313/// use std::fs::File;
314/// use freeswitch_sofia_trace_parser::FrameIterator;
315///
316/// let file = File::open("profile.dump").unwrap();
317/// for frame in FrameIterator::new(file) {
318///     let frame = frame.unwrap();
319///     println!("{} {} bytes {} {}",
320///         frame.timestamp, frame.byte_count, frame.direction, frame.address);
321/// }
322/// ```
323pub struct FrameIterator<R> {
324    reader: R,
325    buf: Vec<u8>,
326    eof: bool,
327    frame_count: u64,
328    offset: u64,
329    stats: ParseStats,
330    skip_tracking: SkipTracking,
331}
332
333impl<R: Read> FrameIterator<R> {
334    /// Create a new frame iterator reading from the given source.
335    pub fn new(reader: R) -> Self {
336        FrameIterator {
337            reader,
338            buf: Vec::with_capacity(READ_BUF_SIZE * 2),
339            eof: false,
340            frame_count: 0,
341            offset: 0,
342            stats: ParseStats::default(),
343            skip_tracking: SkipTracking::CountOnly,
344        }
345    }
346
347    /// Enable capturing of skipped bytes (shorthand for
348    /// [`SkipTracking::CaptureData`]); `false` selects
349    /// [`SkipTracking::CountOnly`]. Whichever of this and
350    /// [`skip_tracking`](Self::skip_tracking) is called last wins.
351    pub fn capture_skipped(mut self, enable: bool) -> Self {
352        self.skip_tracking = if enable {
353            SkipTracking::CaptureData
354        } else {
355            SkipTracking::CountOnly
356        };
357        self
358    }
359
360    /// Set the level of detail for unparsed region tracking.
361    pub fn skip_tracking(mut self, tracking: SkipTracking) -> Self {
362        self.skip_tracking = tracking;
363        self
364    }
365
366    /// Borrow the accumulated parse statistics.
367    pub fn stats(&self) -> &ParseStats {
368        &self.stats
369    }
370
371    /// Take all accumulated unparsed regions, leaving the list empty.
372    pub fn drain_unparsed(&mut self) -> Vec<UnparsedRegion> {
373        self.stats.drain_regions()
374    }
375
376    fn consume(&mut self, n: usize) {
377        self.buf.drain(..n);
378        self.offset += n as u64;
379    }
380
381    fn consume_skipped(&mut self, n: usize, reason: SkipReason) {
382        if self.skip_tracking != SkipTracking::CountOnly {
383            let data = if self.skip_tracking == SkipTracking::CaptureData {
384                Some(self.buf[..n].to_vec())
385            } else {
386                None
387            };
388            self.stats.unparsed_regions.push(UnparsedRegion {
389                offset: self.offset,
390                length: n as u64,
391                reason,
392                data,
393            });
394        }
395        self.stats.bytes_skipped += n as u64;
396        self.consume(n);
397    }
398
399    fn fill_buf(&mut self) -> Result<bool, std::io::Error> {
400        if self.eof {
401            return Ok(false);
402        }
403        let old_len = self.buf.len();
404        self.buf.resize(old_len + READ_BUF_SIZE, 0);
405        let n = self.reader.read(&mut self.buf[old_len..])?;
406        self.buf.truncate(old_len + n);
407        if n == 0 {
408            self.eof = true;
409            return Ok(false);
410        }
411        self.stats.bytes_read += n as u64;
412        Ok(true)
413    }
414
415    /// A SIP header terminator right before a frame boundary is the tail of a
416    /// frame logrotate copied into both files.
417    fn is_replay(&self, skipped: &[u8]) -> bool {
418        if self.frame_count == 0 {
419            return false;
420        }
421        skipped.ends_with(b"\r\n\r\n\x0B\n")
422    }
423
424    /// Find the next `\x0B\n` boundary that is followed by a valid frame header.
425    fn find_boundary(&self, start: usize) -> Option<usize> {
426        let mut search_from = start;
427        loop {
428            let pos = BOUNDARY.find(&self.buf[search_from..])?;
429            let abs_pos = search_from + pos;
430            let after = abs_pos + 2;
431            if after >= self.buf.len() {
432                // Boundary at very end — could be real, but we can't validate header yet
433                // If EOF, accept it as boundary (content ends at \x0B)
434                if self.eof {
435                    return Some(abs_pos);
436                }
437                return None; // Need more data
438            }
439            if is_frame_header(&self.buf[after..]) {
440                return Some(abs_pos);
441            }
442            // \x0B\n in content, not a boundary — skip past it
443            trace!(
444                offset = abs_pos,
445                "found \\x0B\\n in content (not a boundary), skipping"
446            );
447            search_from = abs_pos + 2;
448        }
449    }
450
451    /// Skip to the first valid frame header in the buffer (for partial first frames).
452    fn skip_to_first_header(&mut self) -> Option<usize> {
453        if is_frame_header(&self.buf) {
454            return Some(0);
455        }
456        // Look for \x0B\n followed by a valid header
457        let mut search_from = 0;
458        loop {
459            let pos = BOUNDARY.find(&self.buf[search_from..])?;
460            let abs_pos = search_from + pos;
461            let after = abs_pos + 2;
462            if after < self.buf.len() && is_frame_header(&self.buf[after..]) {
463                info!(skipped_bytes = after, "skipped partial first frame");
464                return Some(after);
465            }
466            search_from = abs_pos + 2;
467        }
468    }
469
470    /// Drop a partial first frame so the buffer starts at a frame header.
471    fn sync_to_first_header(&mut self) -> SyncStep {
472        loop {
473            match self.skip_to_first_header() {
474                Some(offset) => {
475                    if offset > 0 {
476                        let reason = if offset <= MAX_PARTIAL_FRAME {
477                            SkipReason::PartialFirstFrame
478                        } else {
479                            SkipReason::OversizedFrame
480                        };
481                        self.consume_skipped(offset, reason);
482                    }
483                    return SyncStep::Ready;
484                }
485                None => {
486                    if self.eof {
487                        debug!("no valid frame header found in entire input");
488                        let remaining = self.buf.len();
489                        if remaining > 0 {
490                            self.consume_skipped(remaining, SkipReason::InvalidHeader);
491                        }
492                        return SyncStep::End;
493                    }
494                    if let Err(e) = self.fill_buf() {
495                        return SyncStep::Failed(ParseError::Io(e));
496                    }
497                }
498            }
499        }
500    }
501
502    /// Drop `\n` and `\r\n` padding between frames.
503    fn strip_padding(&mut self) {
504        let mut strip = 0;
505        while strip < self.buf.len() {
506            if self.buf[strip] == b'\n' {
507                strip += 1;
508            } else if strip + 1 < self.buf.len()
509                && self.buf[strip] == b'\r'
510                && self.buf[strip + 1] == b'\n'
511            {
512                strip += 2;
513            } else {
514                break;
515            }
516        }
517        if strip > 0 {
518            self.consume(strip);
519        }
520    }
521
522    /// Parse the header at the head of the buffer, reading more as it needs.
523    fn read_header(&mut self) -> HeaderStep {
524        loop {
525            match classify_header(&self.buf) {
526                HeaderParse::Ok(h) => return HeaderStep::Got(h),
527                HeaderParse::NeedMore => {
528                    if self.eof {
529                        debug!("truncated frame header at EOF");
530                        let remaining = self.buf.len();
531                        if remaining > 0 {
532                            self.consume_skipped(remaining, SkipReason::InvalidHeader);
533                        }
534                        return HeaderStep::End;
535                    }
536                    if let Err(e) = self.fill_buf() {
537                        return HeaderStep::Failed(ParseError::Io(e));
538                    }
539                }
540                HeaderParse::Invalid(e) => {
541                    let header_preview: String = self
542                        .buf
543                        .iter()
544                        .take(200)
545                        .take_while(|&&b| b != b'\n')
546                        .map(|&b| {
547                            if b.is_ascii_graphic() || b == b' ' {
548                                b as char
549                            } else {
550                                '.'
551                            }
552                        })
553                        .collect();
554                    if header_preview.starts_with("dump started at ") {
555                        let skip = memchr::memchr(b'\n', &self.buf)
556                            .map(|p| {
557                                let mut end = p + 1;
558                                while end < self.buf.len() && self.buf[end] == b'\n' {
559                                    end += 1;
560                                }
561                                end
562                            })
563                            .unwrap_or(self.buf.len());
564                        debug!(
565                            header = %header_preview,
566                            skipped_bytes = skip,
567                            "skipped dump restart marker",
568                        );
569                        self.consume(skip);
570                        return HeaderStep::Restart;
571                    }
572                    let skip = if let Some(b) = self.find_boundary(0) {
573                        b + 2
574                    } else {
575                        memchr::memchr(b'\n', &self.buf)
576                            .map(|p| p + 1)
577                            .unwrap_or(self.buf.len())
578                    };
579                    let reason = self.classify_skip(skip);
580                    self.consume_skipped(skip, reason);
581                    return HeaderStep::Failed(e);
582                }
583            }
584        }
585    }
586
587    /// Reason for dropping `skip` bytes that failed header parsing.
588    fn classify_skip(&self, skip: usize) -> SkipReason {
589        if self.buf.starts_with(RECV_PREFIX) || self.buf.starts_with(SENT_PREFIX) {
590            SkipReason::InvalidHeader
591        } else if skip > MAX_PARTIAL_FRAME {
592            SkipReason::OversizedFrame
593        } else if self.frame_count == 0 {
594            SkipReason::PartialFirstFrame
595        } else if self.is_replay(&self.buf[..skip]) {
596            SkipReason::ReplayedFrame
597        } else {
598            SkipReason::MidStreamSkip
599        }
600    }
601
602    /// Read one frame's content, from the header's end to the frame boundary.
603    fn read_content(&mut self, header: FrameHeader) -> Option<Result<Frame, ParseError>> {
604        let FrameHeader {
605            direction,
606            byte_count,
607            transport,
608            address,
609            timestamp,
610            header_len,
611        } = header;
612        let content_start = header_len;
613        let expected_end = content_start + byte_count;
614
615        // Find the boundary for this frame.
616        // Strategy: first check at the expected position (content_start + byte_count),
617        // then fall back to scanning. This handles file concatenation where \x0B\n
618        // is followed by garbage from the next file's truncated first frame.
619        let content = loop {
620            // Ensure we have enough data to check the expected position, but
621            // never buffer a declared count further than a frame can reach:
622            // past that, boundary scanning takes over.
623            let fill_to = (expected_end + 1).min(content_start + MAX_PARTIAL_FRAME);
624            while self.buf.len() <= fill_to && !self.eof {
625                if let Err(e) = self.fill_buf() {
626                    return Some(Err(ParseError::Io(e)));
627                }
628            }
629
630            // Check at expected position first (byte_count hint)
631            if expected_end < self.buf.len() && self.buf[expected_end] == 0x0B {
632                let has_newline =
633                    expected_end + 1 < self.buf.len() && self.buf[expected_end + 1] == b'\n';
634                let at_eof = expected_end + 1 >= self.buf.len() && self.eof;
635
636                if has_newline || at_eof {
637                    let content = self.buf[content_start..expected_end].to_vec();
638                    let drain_to = if has_newline {
639                        expected_end + 2
640                    } else {
641                        expected_end + 1
642                    };
643                    self.consume(drain_to);
644                    self.frame_count += 1;
645                    break content;
646                }
647            }
648
649            // Fall back to scanning for \x0B\n + valid header
650            if let Some(boundary_pos) = self.find_boundary(content_start) {
651                let content = self.buf[content_start..boundary_pos].to_vec();
652                let drain_to = boundary_pos + 2;
653                self.consume(drain_to);
654                self.frame_count += 1;
655
656                if content.len() != byte_count {
657                    debug!(
658                        frame = self.frame_count,
659                        expected = byte_count,
660                        actual = content.len(),
661                        "frame content size mismatch"
662                    );
663                }
664
665                break content;
666            }
667
668            if self.eof {
669                // Last frame — no trailing \x0B\n
670                let end = if self.buf.last() == Some(&0x0B) {
671                    self.buf.len() - 1
672                } else {
673                    self.buf.len()
674                };
675                let content = self.buf[content_start..end].to_vec();
676                let len = self.buf.len();
677                self.consume(len);
678                self.frame_count += 1;
679
680                if content.len() < byte_count {
681                    let missing = byte_count - content.len();
682                    warn!(
683                        frame = self.frame_count,
684                        expected = byte_count,
685                        actual = content.len(),
686                        missing,
687                        "incomplete frame at EOF"
688                    );
689                    if self.skip_tracking != SkipTracking::CountOnly {
690                        self.stats.unparsed_regions.push(UnparsedRegion {
691                            offset: self.offset,
692                            length: missing as u64,
693                            reason: SkipReason::IncompleteFrame,
694                            data: None,
695                        });
696                    }
697                } else if content.len() != byte_count {
698                    debug!(
699                        frame = self.frame_count,
700                        expected = byte_count,
701                        actual = content.len(),
702                        "last frame content size mismatch"
703                    );
704                }
705
706                break content;
707            }
708
709            if let Err(e) = self.fill_buf() {
710                return Some(Err(ParseError::Io(e)));
711            }
712        };
713
714        Some(Ok(Frame {
715            direction,
716            byte_count,
717            transport,
718            address,
719            timestamp,
720            content,
721        }))
722    }
723}
724
725impl<R: Read> Iterator for FrameIterator<R> {
726    type Item = Result<Frame, ParseError>;
727
728    fn next(&mut self) -> Option<Self::Item> {
729        let header = loop {
730            if self.buf.is_empty() && !self.eof {
731                if let Err(e) = self.fill_buf() {
732                    return Some(Err(ParseError::Io(e)));
733                }
734            }
735            if self.buf.is_empty() {
736                return None;
737            }
738
739            if self.frame_count == 0 {
740                match self.sync_to_first_header() {
741                    SyncStep::Ready => {}
742                    SyncStep::End => return None,
743                    SyncStep::Failed(e) => return Some(Err(e)),
744                }
745            }
746
747            self.strip_padding();
748            if self.buf.is_empty() {
749                continue;
750            }
751
752            match self.read_header() {
753                HeaderStep::Got(header) => break header,
754                HeaderStep::Restart => continue,
755                HeaderStep::End => return None,
756                HeaderStep::Failed(e) => return Some(Err(e)),
757            }
758        };
759
760        self.read_content(header)
761    }
762}
763
764#[cfg(test)]
765mod tests {
766    use super::*;
767    use crate::types::SkipTracking;
768
769    #[test]
770    fn parse_recv_ipv4_tcp() {
771        let header = b"recv 100 bytes from tcp/192.168.1.1:5060 at 00:00:01.350874:\n";
772        let h = parse_frame_header(header).unwrap();
773        assert_eq!(h.direction, Direction::Recv);
774        assert_eq!(h.byte_count, 100);
775        assert_eq!(h.transport, Transport::Tcp);
776        assert_eq!(h.address, "192.168.1.1:5060");
777        assert_eq!(
778            h.timestamp,
779            Timestamp::TimeOnly {
780                hour: 0,
781                min: 0,
782                sec: 1,
783                usec: 350874
784            }
785        );
786        assert_eq!(h.header_len, header.len());
787    }
788
789    #[test]
790    fn parse_recv_ipv6_tcp() {
791        let header = b"recv 1440 bytes from tcp/[2001:4958:10:14::4]:30046 at 13:03:21.674883:\n";
792        let h = parse_frame_header(header).unwrap();
793        assert_eq!(h.direction, Direction::Recv);
794        assert_eq!(h.byte_count, 1440);
795        assert_eq!(h.transport, Transport::Tcp);
796        assert_eq!(h.address, "[2001:4958:10:14::4]:30046");
797        assert_eq!(
798            h.timestamp,
799            Timestamp::TimeOnly {
800                hour: 13,
801                min: 3,
802                sec: 21,
803                usec: 674883
804            }
805        );
806    }
807
808    #[test]
809    fn parse_sent_ipv6_tcp() {
810        let header = b"sent 681 bytes to tcp/[2001:4958:10:14::4]:30046 at 13:03:21.675500:\n";
811        let h = parse_frame_header(header).unwrap();
812        assert_eq!(h.direction, Direction::Sent);
813        assert_eq!(h.byte_count, 681);
814        assert_eq!(h.transport, Transport::Tcp);
815        assert_eq!(h.address, "[2001:4958:10:14::4]:30046");
816    }
817
818    #[test]
819    fn parse_recv_udp() {
820        let header = b"recv 457 bytes from udp/10.0.0.1:5060 at 00:19:47.123456:\n";
821        let h = parse_frame_header(header).unwrap();
822        assert_eq!(h.direction, Direction::Recv);
823        assert_eq!(h.transport, Transport::Udp);
824    }
825
826    #[test]
827    fn parse_sent_tls() {
828        let header = b"sent 500 bytes to tls/10.0.0.1:5061 at 12:00:00.000000:\n";
829        let h = parse_frame_header(header).unwrap();
830        assert_eq!(h.direction, Direction::Sent);
831        assert_eq!(h.byte_count, 500);
832        assert_eq!(h.transport, Transport::Tls);
833    }
834
835    #[test]
836    fn parse_full_datetime_timestamp() {
837        let header = b"recv 100 bytes from tcp/192.168.1.1:5060 at 2026-02-01 10:00:00.000000:\n";
838        let h = parse_frame_header(header).unwrap();
839        assert_eq!(
840            h.timestamp,
841            Timestamp::DateTime {
842                year: 2026,
843                month: 2,
844                day: 1,
845                hour: 10,
846                min: 0,
847                sec: 0,
848                usec: 0
849            }
850        );
851    }
852
853    #[test]
854    fn parse_invalid_header() {
855        assert!(parse_frame_header(b"invalid header\n").is_err());
856        assert!(
857            parse_frame_header(b"recv abc bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\n")
858                .is_err()
859        );
860    }
861
862    #[test]
863    fn is_frame_header_valid() {
864        assert!(is_frame_header(
865            b"recv 100 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\n"
866        ));
867        assert!(is_frame_header(
868            b"sent 681 bytes to tcp/[::1]:5060 at 00:00:00.000000:\n"
869        ));
870        assert!(!is_frame_header(b"not a header"));
871        assert!(!is_frame_header(b"recv abc bytes"));
872        assert!(!is_frame_header(b""));
873    }
874
875    #[test]
876    fn frame_iterator_single_frame() {
877        let data = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n";
878        let frames: Vec<Frame> = FrameIterator::new(&data[..])
879            .collect::<Result<Vec<_>, _>>()
880            .unwrap();
881        assert_eq!(frames.len(), 1);
882        assert_eq!(frames[0].content, b"hello");
883        assert_eq!(frames[0].byte_count, 5);
884    }
885
886    #[test]
887    fn frame_iterator_multiple_frames() {
888        let data = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\nsent 5 bytes to tcp/1.1.1.1:5060 at 00:00:00.000001:\nworld\x0B\n";
889        let frames: Vec<Frame> = FrameIterator::new(&data[..])
890            .collect::<Result<Vec<_>, _>>()
891            .unwrap();
892        assert_eq!(frames.len(), 2);
893        assert_eq!(frames[0].content, b"hello");
894        assert_eq!(frames[0].direction, Direction::Recv);
895        assert_eq!(frames[1].content, b"world");
896        assert_eq!(frames[1].direction, Direction::Sent);
897    }
898
899    #[test]
900    fn frame_iterator_vt_in_content() {
901        // \x0B in content but not followed by valid header — should NOT split
902        let mut data = Vec::new();
903        data.extend_from_slice(b"recv 15 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\n");
904        data.extend_from_slice(b"he\x0B\nllo world!!");
905        data.extend_from_slice(b"\x0B\n");
906        let frames: Vec<Frame> = FrameIterator::new(&data[..])
907            .collect::<Result<Vec<_>, _>>()
908            .unwrap();
909        assert_eq!(frames.len(), 1);
910        assert_eq!(frames[0].content, b"he\x0B\nllo world!!");
911    }
912
913    #[test]
914    fn frame_iterator_eof_without_boundary() {
915        let data = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello";
916        let frames: Vec<Frame> = FrameIterator::new(&data[..])
917            .collect::<Result<Vec<_>, _>>()
918            .unwrap();
919        assert_eq!(frames.len(), 1);
920        assert_eq!(frames[0].content, b"hello");
921    }
922
923    #[test]
924    fn frame_iterator_eof_with_lone_vt() {
925        let data = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B";
926        let frames: Vec<Frame> = FrameIterator::new(&data[..])
927            .collect::<Result<Vec<_>, _>>()
928            .unwrap();
929        assert_eq!(frames.len(), 1);
930        assert_eq!(frames[0].content, b"hello");
931    }
932
933    #[test]
934    fn frame_iterator_partial_first_frame() {
935        // Data starts with garbage, then a valid boundary + frame
936        let mut data = Vec::new();
937        data.extend_from_slice(b"partial garbage data");
938        data.extend_from_slice(b"\x0B\n");
939        data.extend_from_slice(
940            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
941        );
942        let frames: Vec<Frame> = FrameIterator::new(&data[..])
943            .collect::<Result<Vec<_>, _>>()
944            .unwrap();
945        assert_eq!(frames.len(), 1);
946        assert_eq!(frames[0].content, b"hello");
947    }
948
949    #[test]
950    fn frame_iterator_truncated_last_frame() {
951        // Complete frame followed by truncated frame at EOF (no \x0B\n)
952        let mut data = Vec::new();
953        data.extend_from_slice(
954            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
955        );
956        data.extend_from_slice(b"sent 3 bytes to tcp/1.1.1.1:5060 at 00:00:01.000000:\nbye");
957        let frames: Vec<Frame> = FrameIterator::new(&data[..])
958            .collect::<Result<Vec<_>, _>>()
959            .unwrap();
960        assert_eq!(frames.len(), 2);
961        assert_eq!(frames[0].content, b"hello");
962        assert_eq!(frames[1].content, b"bye");
963    }
964
965    #[test]
966    fn frame_iterator_file_concatenation() {
967        // Simulates `cat dump.20 dump.21 | parser`
968        // File 1: truncated start + valid frame + complete end
969        // File 2: truncated start (no header) + valid frame
970        let mut data = Vec::new();
971
972        // File 1: starts with valid header
973        data.extend_from_slice(
974            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
975        );
976        data.extend_from_slice(
977            b"sent 5 bytes to tcp/1.1.1.1:5060 at 00:00:00.000001:\nworld\x0B\n",
978        );
979
980        // File 2: starts with truncated frame data (no header), then boundary, then valid frame
981        data.extend_from_slice(b"some truncated SIP content from previous rotation\r\n\r\n");
982        data.extend_from_slice(b"\x0B\n");
983        data.extend_from_slice(
984            b"recv 3 bytes from tcp/2.2.2.2:5060 at 01:00:00.000000:\nfoo\x0B\n",
985        );
986
987        let items: Vec<Result<Frame, ParseError>> = FrameIterator::new(&data[..]).collect();
988        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
989        assert_eq!(frames.len(), 3);
990        assert_eq!(frames[0].content, b"hello");
991        assert_eq!(frames[1].content, b"world");
992        assert_eq!(frames[2].content, b"foo");
993        assert_eq!(frames[2].address, "2.2.2.2:5060");
994    }
995
996    #[test]
997    fn frame_iterator_file_concatenation_mid_stream_garbage() {
998        // The join point between files produces garbage that looks like:
999        // ...last_content\x0B\ntruncated_first_of_next_file\x0B\nvalid_header...
1000        // The truncated part is NOT a valid header, so recovery should skip it
1001        let mut data = Vec::new();
1002
1003        // Last frame of file 1
1004        data.extend_from_slice(
1005            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1006        );
1007
1008        // Truncated first frame of file 2 (mid-SIP content, no frame header)
1009        data.extend_from_slice(b"Content-Type: application/sdp\r\n\r\nv=0\r\n");
1010        data.extend_from_slice(b"\x0B\n");
1011
1012        // Valid second frame of file 2
1013        data.extend_from_slice(b"sent 3 bytes to tcp/3.3.3.3:5060 at 02:00:00.000000:\nbar\x0B\n");
1014
1015        let items: Vec<Result<Frame, ParseError>> = FrameIterator::new(&data[..]).collect();
1016        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1017        assert_eq!(frames.len(), 2);
1018        assert_eq!(frames[0].content, b"hello");
1019        assert_eq!(frames[1].content, b"bar");
1020    }
1021
1022    #[test]
1023    fn frame_iterator_empty_input() {
1024        let data: &[u8] = b"";
1025        let frames: Vec<Result<Frame, ParseError>> = FrameIterator::new(data).collect();
1026        assert!(frames.is_empty());
1027    }
1028
1029    #[test]
1030    fn frame_iterator_only_garbage() {
1031        let data = b"this is not a SIP trace dump at all, just garbage text";
1032        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1033        let frames: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1034        assert!(frames.is_empty());
1035        let stats = iter.stats();
1036        assert_eq!(stats.bytes_read, data.len() as u64);
1037        assert_eq!(stats.bytes_skipped, data.len() as u64);
1038        assert_eq!(stats.unparsed_regions.len(), 1);
1039        assert_eq!(stats.unparsed_regions[0].reason, SkipReason::InvalidHeader);
1040    }
1041
1042    #[test]
1043    fn frame_iterator_truncated_header_at_eof() {
1044        let data = b"recv 5 bytes from tcp/1.1.1.1:5060";
1045        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1046        let frames: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1047        assert!(frames.is_empty());
1048        let stats = iter.stats();
1049        assert_eq!(stats.bytes_read, data.len() as u64);
1050        assert_eq!(stats.bytes_skipped, data.len() as u64);
1051    }
1052
1053    #[test]
1054    fn frame_iterator_dump_marker_at_eof() {
1055        // A dump restart marker at the end of input (with trailing \n\n as in real dumps)
1056        // should be silently consumed, not returned as an error.
1057        let mut data = Vec::new();
1058        data.extend_from_slice(
1059            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1060        );
1061        data.extend_from_slice(b"dump started at Thu Aug 22 11:38:11 2024\n\n\n");
1062
1063        let frames: Vec<Result<Frame, ParseError>> = FrameIterator::new(&data[..]).collect();
1064        assert_eq!(frames.len(), 1);
1065        assert!(frames[0].is_ok());
1066        assert_eq!(frames[0].as_ref().unwrap().content, b"hello");
1067    }
1068
1069    #[test]
1070    fn frame_iterator_dump_marker_mid_stream() {
1071        // A dump restart marker between two valid frames (with trailing \n\n as in real dumps)
1072        // should be skipped, and both frames should parse successfully.
1073        let mut data = Vec::new();
1074        data.extend_from_slice(
1075            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1076        );
1077        data.extend_from_slice(b"dump started at Thu Aug 22 11:38:11 2024\n\n\n");
1078        data.extend_from_slice(b"sent 3 bytes to tcp/2.2.2.2:5060 at 00:00:01.000000:\nbye\x0B\n");
1079
1080        let frames: Vec<Result<Frame, ParseError>> = FrameIterator::new(&data[..]).collect();
1081        assert_eq!(frames.len(), 2);
1082        assert_eq!(frames[0].as_ref().unwrap().content, b"hello");
1083        assert_eq!(frames[1].as_ref().unwrap().content, b"bye");
1084    }
1085
1086    #[test]
1087    fn frame_iterator_multiple_newlines_after_boundary() {
1088        // Multiple \n and \r\n between frames should all be stripped.
1089        let mut data = Vec::new();
1090        data.extend_from_slice(
1091            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1092        );
1093        data.extend_from_slice(b"\n\r\n\n");
1094        data.extend_from_slice(
1095            b"sent 5 bytes to tcp/1.1.1.1:5060 at 00:00:00.000001:\nworld\x0B\n",
1096        );
1097
1098        let frames: Vec<Frame> = FrameIterator::new(&data[..])
1099            .collect::<Result<Vec<_>, _>>()
1100            .unwrap();
1101        assert_eq!(frames.len(), 2);
1102        assert_eq!(frames[0].content, b"hello");
1103        assert_eq!(frames[1].content, b"world");
1104    }
1105
1106    #[test]
1107    fn stats_clean_input() {
1108        let data = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n";
1109        let mut iter = FrameIterator::new(&data[..]);
1110        let frames: Vec<Frame> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1111        assert_eq!(frames.len(), 1);
1112        let stats = iter.stats();
1113        assert_eq!(stats.bytes_read, data.len() as u64);
1114        assert_eq!(stats.bytes_skipped, 0);
1115        assert!(stats.unparsed_regions.is_empty());
1116    }
1117
1118    #[test]
1119    fn stats_multiple_frames() {
1120        let data = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\nsent 5 bytes to tcp/1.1.1.1:5060 at 00:00:00.000001:\nworld\x0B\n";
1121        let mut iter = FrameIterator::new(&data[..]);
1122        let frames: Vec<Frame> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1123        assert_eq!(frames.len(), 2);
1124        let stats = iter.stats();
1125        assert_eq!(stats.bytes_read, data.len() as u64);
1126        assert_eq!(stats.bytes_skipped, 0);
1127        assert!(stats.unparsed_regions.is_empty());
1128    }
1129
1130    #[test]
1131    fn stats_partial_first_frame() {
1132        let mut data = Vec::new();
1133        data.extend_from_slice(b"partial garbage data");
1134        data.extend_from_slice(b"\x0B\n");
1135        data.extend_from_slice(
1136            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1137        );
1138        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1139        let frames: Vec<Frame> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1140        assert_eq!(frames.len(), 1);
1141        let stats = iter.stats();
1142        assert_eq!(stats.bytes_read, data.len() as u64);
1143        // "partial garbage data" + "\x0B\n" = 21 bytes skipped
1144        let skipped = b"partial garbage data\x0B\n".len() as u64;
1145        assert_eq!(stats.bytes_skipped, skipped);
1146        assert_eq!(stats.unparsed_regions.len(), 1);
1147        assert_eq!(stats.unparsed_regions[0].offset, 0);
1148        assert_eq!(stats.unparsed_regions[0].length, skipped);
1149        assert_eq!(
1150            stats.unparsed_regions[0].reason,
1151            crate::types::SkipReason::PartialFirstFrame
1152        );
1153        assert!(stats.unparsed_regions[0].data.is_none());
1154    }
1155
1156    #[test]
1157    fn stats_partial_first_frame_capture() {
1158        let mut data = Vec::new();
1159        data.extend_from_slice(b"partial garbage data");
1160        data.extend_from_slice(b"\x0B\n");
1161        data.extend_from_slice(
1162            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1163        );
1164        let mut iter = FrameIterator::new(&data[..]).capture_skipped(true);
1165        let frames: Vec<Frame> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1166        assert_eq!(frames.len(), 1);
1167        let stats = iter.stats();
1168        assert_eq!(stats.unparsed_regions.len(), 1);
1169        let region = &stats.unparsed_regions[0];
1170        assert_eq!(
1171            region.data.as_deref(),
1172            Some(b"partial garbage data\x0B\n".as_slice())
1173        );
1174    }
1175
1176    #[test]
1177    fn stats_mid_stream_partial_frame() {
1178        // SIP content between valid frames (file concatenation scenario)
1179        let mut data = Vec::new();
1180        data.extend_from_slice(
1181            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1182        );
1183        data.extend_from_slice(b"Content-Type: application/sdp\r\n\r\nv=0\r\n");
1184        data.extend_from_slice(b"\x0B\n");
1185        data.extend_from_slice(b"sent 3 bytes to tcp/3.3.3.3:5060 at 02:00:00.000000:\nbar\x0B\n");
1186
1187        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1188        let items: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1189        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1190        assert_eq!(frames.len(), 2);
1191        let stats = iter.stats();
1192        assert!(stats.bytes_skipped > 0);
1193        assert_eq!(stats.unparsed_regions.len(), 1);
1194        assert_eq!(
1195            stats.unparsed_regions[0].reason,
1196            crate::types::SkipReason::MidStreamSkip
1197        );
1198    }
1199
1200    #[test]
1201    fn stats_replayed_frame() {
1202        // Simulate logrotate: a frame's tail (SIP headers ending with \r\n\r\n\x0B\n)
1203        // appears between two valid frames at a file boundary.
1204        let frame1 = b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n";
1205        let replay = b"Route: <sip:10.0.0.1:5060;lr>\r\nContent-Length: 0\r\n\r\n\x0B\n";
1206        let frame2 = b"sent 3 bytes to tcp/3.3.3.3:5060 at 02:00:00.000000:\nbar\x0B\n";
1207
1208        let mut data = Vec::new();
1209        data.extend_from_slice(frame1);
1210        data.extend_from_slice(replay);
1211        data.extend_from_slice(frame2);
1212
1213        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1214        let items: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1215        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1216        assert_eq!(frames.len(), 2);
1217        let stats = iter.stats();
1218        assert_eq!(stats.unparsed_regions.len(), 1);
1219        assert_eq!(
1220            stats.unparsed_regions[0].reason,
1221            crate::types::SkipReason::ReplayedFrame
1222        );
1223    }
1224
1225    #[test]
1226    fn stats_incomplete_frame_at_eof() {
1227        // Frame header says 100 bytes but only 20 bytes available before EOF
1228        let mut data = Vec::new();
1229        data.extend_from_slice(
1230            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1231        );
1232        data.extend_from_slice(b"recv 100 bytes from tcp/2.2.2.2:5060 at 01:00:00.000000:\n");
1233        data.extend_from_slice(b"partial content only");
1234        // No \x0B\n boundary — EOF truncation
1235
1236        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1237        let items: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1238        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1239        assert_eq!(frames.len(), 2, "truncated frame should still be returned");
1240        assert_eq!(frames[1].content, b"partial content only");
1241        assert_eq!(frames[1].byte_count, 100);
1242        let stats = iter.stats();
1243        assert_eq!(stats.unparsed_regions.len(), 1);
1244        assert_eq!(
1245            stats.unparsed_regions[0].reason,
1246            crate::types::SkipReason::IncompleteFrame
1247        );
1248    }
1249
1250    #[test]
1251    fn stats_invalid_header_skip() {
1252        // Malformed frame header (starts with recv/sent but unparseable)
1253        let mut data = Vec::new();
1254        data.extend_from_slice(
1255            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1256        );
1257        data.extend_from_slice(b"recv CORRUPT HEADER garbage\n");
1258        data.extend_from_slice(b"\x0B\n");
1259        data.extend_from_slice(b"sent 3 bytes to tcp/3.3.3.3:5060 at 02:00:00.000000:\nbar\x0B\n");
1260
1261        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1262        let items: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1263        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1264        assert_eq!(frames.len(), 2);
1265        let stats = iter.stats();
1266        assert!(stats.bytes_skipped > 0);
1267        assert_eq!(stats.unparsed_regions.len(), 1);
1268        assert_eq!(
1269            stats.unparsed_regions[0].reason,
1270            crate::types::SkipReason::InvalidHeader
1271        );
1272    }
1273
1274    #[test]
1275    fn stats_oversized_frame_at_start() {
1276        let mut data = Vec::new();
1277        data.resize(MAX_PARTIAL_FRAME + 1, b'x');
1278        data.extend_from_slice(b"\x0B\n");
1279        data.extend_from_slice(
1280            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1281        );
1282        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1283        let items: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1284        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1285        assert_eq!(frames.len(), 1);
1286        let stats = iter.stats();
1287        assert_eq!(stats.unparsed_regions.len(), 1);
1288        assert_eq!(
1289            stats.unparsed_regions[0].reason,
1290            crate::types::SkipReason::OversizedFrame
1291        );
1292    }
1293
1294    #[test]
1295    fn stats_oversized_frame_mid_stream() {
1296        let mut data = Vec::new();
1297        data.extend_from_slice(
1298            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1299        );
1300        let garbage_len = MAX_PARTIAL_FRAME + 1;
1301        data.resize(data.len() + garbage_len, b'x');
1302        data.extend_from_slice(b"\x0B\n");
1303        data.extend_from_slice(b"sent 3 bytes to tcp/3.3.3.3:5060 at 02:00:00.000000:\nbar\x0B\n");
1304        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1305        let items: Vec<Result<Frame, ParseError>> = iter.by_ref().collect();
1306        let frames: Vec<Frame> = items.into_iter().filter_map(Result::ok).collect();
1307        assert_eq!(frames.len(), 2);
1308        let stats = iter.stats();
1309        assert_eq!(stats.unparsed_regions.len(), 1);
1310        assert_eq!(
1311            stats.unparsed_regions[0].reason,
1312            crate::types::SkipReason::OversizedFrame
1313        );
1314    }
1315
1316    #[test]
1317    fn overlong_byte_count_does_not_buffer_ahead() {
1318        let mut data = Vec::new();
1319        data.extend_from_slice(
1320            format!(
1321                "recv {} bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\n",
1322                MAX_PARTIAL_FRAME * 10
1323            )
1324            .as_bytes(),
1325        );
1326        data.extend_from_slice(b"hello\x0B\n");
1327        while data.len() < MAX_PARTIAL_FRAME * 10 {
1328            data.extend_from_slice(
1329                b"sent 3 bytes to tcp/3.3.3.3:5060 at 02:00:00.000000:\nbar\x0B\n",
1330            );
1331        }
1332
1333        let mut iter = FrameIterator::new(&data[..]);
1334        let first = iter.next().unwrap().unwrap();
1335        assert_eq!(first.content, b"hello");
1336        assert!(
1337            iter.buf.len() <= MAX_PARTIAL_FRAME + READ_BUF_SIZE,
1338            "buffered {} bytes on a byte_count ten times the frame",
1339            iter.buf.len()
1340        );
1341        let second = iter.next().unwrap().unwrap();
1342        assert_eq!(second.content, b"bar");
1343    }
1344
1345    #[test]
1346    fn stats_partial_first_frame_within_limit() {
1347        // Content + \x0B\n boundary = MAX_PARTIAL_FRAME, should still be PartialFirstFrame
1348        let mut data = Vec::new();
1349        data.resize(MAX_PARTIAL_FRAME - 2, b'x');
1350        data.extend_from_slice(b"\x0B\n");
1351        data.extend_from_slice(
1352            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1353        );
1354        let mut iter = FrameIterator::new(&data[..]).skip_tracking(SkipTracking::TrackRegions);
1355        let frames: Vec<Frame> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1356        assert_eq!(frames.len(), 1);
1357        let stats = iter.stats();
1358        assert_eq!(stats.unparsed_regions.len(), 1);
1359        assert_eq!(
1360            stats.unparsed_regions[0].reason,
1361            crate::types::SkipReason::PartialFirstFrame
1362        );
1363    }
1364
1365    #[test]
1366    fn stats_dump_restart_marker() {
1367        let mut data = Vec::new();
1368        data.extend_from_slice(
1369            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1370        );
1371        data.extend_from_slice(b"dump started at Thu Aug 22 11:38:11 2024\n\n\n");
1372        data.extend_from_slice(b"sent 3 bytes to tcp/2.2.2.2:5060 at 00:00:01.000000:\nbye\x0B\n");
1373
1374        let mut iter = FrameIterator::new(&data[..]);
1375        let frames: Vec<Frame> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1376        assert_eq!(frames.len(), 2);
1377        let stats = iter.stats();
1378        // Dump restart marker is structural, not skipped
1379        assert_eq!(stats.bytes_skipped, 0);
1380        assert!(stats.unparsed_regions.is_empty());
1381    }
1382
1383    #[test]
1384    fn stats_count_only_no_regions() {
1385        let mut data = Vec::new();
1386        data.extend_from_slice(b"partial garbage data");
1387        data.extend_from_slice(b"\x0B\n");
1388        data.extend_from_slice(
1389            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1390        );
1391        let mut iter = FrameIterator::new(&data[..]);
1392        let frames: Vec<_> = iter.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
1393        assert_eq!(frames.len(), 1);
1394        let stats = iter.stats();
1395        let skipped = b"partial garbage data\x0B\n".len() as u64;
1396        assert_eq!(stats.bytes_skipped, skipped);
1397        assert!(
1398            stats.unparsed_regions.is_empty(),
1399            "CountOnly should not accumulate regions"
1400        );
1401    }
1402
1403    #[test]
1404    fn frame_iterator_trailing_newlines_at_eof() {
1405        // Trailing newlines after the last boundary at EOF should not cause errors.
1406        let mut data = Vec::new();
1407        data.extend_from_slice(
1408            b"recv 5 bytes from tcp/1.1.1.1:5060 at 00:00:00.000000:\nhello\x0B\n",
1409        );
1410        data.extend_from_slice(b"\n\n");
1411
1412        let frames: Vec<Frame> = FrameIterator::new(&data[..])
1413            .collect::<Result<Vec<_>, _>>()
1414            .unwrap();
1415        assert_eq!(frames.len(), 1);
1416        assert_eq!(frames[0].content, b"hello");
1417    }
1418}
freeswitch_sofia_trace_parser/frame.rs

freeswitch_sofia_trace_parser/
frame.rs
