agent_sessions/reader/
stream.rs

1use super::{ReadStatus, SessionReader, TailMode, input::read_record};
2use crate::parser::tally;
3use crate::statistics::{DecodeError, decode};
4use crate::{Event, LineErrorKind, Located, Location, StreamError};
5use std::io::BufRead;
6
7impl<R: BufRead> Iterator for SessionReader<R> {
8    type Item = Result<Located<Event>, StreamError>;
9    fn next(&mut self) -> Option<Self::Item> {
10        if let Some(event) = self.deliver() {
11            return Some(Ok(event));
12        }
13        if self.ended {
14            return None;
15        }
16        loop {
17            let start = self.summary.bytes_read;
18            let record = match read_record(
19                &mut self.source,
20                &self.opts,
21                &mut self.summary,
22                std::mem::take(&mut self.scratch),
23            ) {
24                Ok(Some(record)) => record,
25                Ok(None) => {
26                    self.ended = true;
27                    self.summary.status = if self.summary.line_errors == 0 {
28                        ReadStatus::Complete
29                    } else {
30                        ReadStatus::CompleteWithErrors
31                    };
32                    return None;
33                }
34                Err(error) => {
35                    self.ended = true;
36                    self.summary.status = ReadStatus::Failed;
37                    return Some(Err(error));
38                }
39            };
40            self.summary.lines += 1;
41            let error = if record.too_long {
42                Some(LineErrorKind::TooLong)
43            } else if std::str::from_utf8(&record.bytes).is_err() {
44                Some(LineErrorKind::InvalidUtf8)
45            } else {
46                None
47            };
48            if let Some(kind) = error {
49                self.scratch = record.bytes;
50                return Some(Err(self.line_error(start, kind)));
51            }
52            if record.bytes.iter().all(u8::is_ascii_whitespace) {
53                self.scratch = record.bytes;
54                self.checkpoint_record();
55                continue;
56            }
57            let decoded = decode(self.agent, &record.bytes, &mut self.state, &self.opts);
58            self.scratch = record.bytes;
59            let mut parsed = match decoded {
60                Ok(parsed) => parsed,
61                Err(DecodeError::Json(e))
62                    if !record.newline
63                        && e.is_eof()
64                        && self.opts.tail == TailMode::AllowIncomplete =>
65                {
66                    self.ended = true;
67                    self.summary.truncated_tail = true;
68                    self.summary.status = ReadStatus::IncompleteTail;
69                    return None;
70                }
71                Err(DecodeError::Json(_)) => {
72                    return Some(Err(self.line_error(start, LineErrorKind::InvalidJson)));
73                }
74                Err(DecodeError::Fields(kind)) => return Some(Err(self.line_error(start, kind))),
75            };
76            if let Some((known, label)) = &parsed.ignored_one {
77                if *known {
78                    tally(&mut self.summary.ignored_types, label);
79                } else {
80                    tally(&mut self.summary.unknown_types, label);
81                }
82            }
83            for unknown in &parsed.unknown {
84                tally(&mut self.summary.unknown_types, unknown);
85            }
86            for ignored in &parsed.ignored {
87                tally(&mut self.summary.ignored_types, ignored);
88            }
89            let event_count = parsed.events.len();
90            for (ordinal, (event_index, mut event)) in parsed.events.into_iter().enumerate() {
91                if let Event::Message(m) = &mut event {
92                    m.is_sidechain |= self.state.sidechain || self.file_sidechain;
93                }
94                if !self.opts.include.includes(&event) {
95                    continue;
96                }
97                self.pending.push_back(Located {
98                    location: Location {
99                        record_index: self.summary.lines - 1,
100                        line_no: self.summary.lines,
101                        byte_start: start,
102                        byte_end: self.summary.bytes_read,
103                        event_index,
104                    },
105                    at: parsed.at,
106                    timestamp_text: if ordinal + 1 == event_count {
107                        parsed.timestamp_text.take()
108                    } else {
109                        parsed.timestamp_text.clone()
110                    },
111                    record_id: if ordinal + 1 == event_count {
112                        parsed.record_id.take()
113                    } else {
114                        parsed.record_id.clone()
115                    },
116                    record_type: if ordinal + 1 == event_count {
117                        parsed.record_type.take()
118                    } else {
119                        parsed.record_type.clone()
120                    },
121                    session_id: self.state.session_id.clone(),
122                    message_id: if ordinal + 1 == event_count {
123                        parsed.message_id.take()
124                    } else {
125                        parsed.message_id.clone()
126                    },
127                    value: event,
128                });
129            }
130            self.checkpoint_record();
131            if let Some(event) = self.deliver() {
132                return Some(Ok(event));
133            }
134        }
135    }
136}
137
138impl<R: BufRead> SessionReader<R> {
139    fn checkpoint_record(&mut self) {
140        if self.summary.line_errors == 0 {
141            self.checkpoint = Some(self.summary.bytes_read);
142            if self.pending.is_empty() {
143                self.summary.last_complete_byte = self.summary.bytes_read;
144                self.checkpoint = None;
145            }
146        }
147    }
148    fn line_error(&mut self, byte_start: u64, kind: LineErrorKind) -> StreamError {
149        self.summary.line_errors += 1;
150        if !matches!(
151            kind,
152            LineErrorKind::CounterRegression | LineErrorKind::LostUsageBaseline
153        ) {
154            self.state.broken_usage = true;
155        }
156        StreamError::Line {
157            line_no: self.summary.lines,
158            byte_start,
159            kind,
160        }
161    }
162}
agent_sessions/reader/stream.rs

agent_sessions/reader/
stream.rs
