freeswitch_sofia_trace_parser/
frame.rs

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

freeswitch_sofia_trace_parser/
frame.rs
