kui_trace/
stream.rs

1//! A recording read as it arrives (backlog F249, ADR 0075 decision 5):
2//! bytes pushed in as a [`Transport`] hands them over — a file still being
3//! written, a pipe, a socket — decoded up to the last whole record, the
4//! cut one kept for the next push, the recording and its timeline grown.
5//!
6//! ```
7//! use kui_trace::{Record, Stream, Trace};
8//!
9//! let mut t = Trace::new();
10//! t.records.push(Record::FrameBegin { t: 0, frame: 1, causes: vec![] });
11//! t.records.push(Record::FrameEnd { t: 900, frame: 1, unchanged: Some(false), nodes: 3 });
12//! let bytes = t.to_bytes();
13//!
14//! let mut s = Stream::new();
15//! let (head, rest) = bytes.split_at(bytes.len() - 2);
16//! s.push(head)?;
17//! assert_eq!(s.timeline().frames[0].end, None, "the end is cut");
18//! s.push(rest)?;
19//! assert_eq!(s.timeline().frames[0].end, Some(900));
20//! # Ok::<(), kui_trace::Error>(())
21//! ```
22
23use std::io;
24
25use crate::codec::{Decoder, Error, check_header, framed};
26use crate::model::{Timeline, Trace};
27
28/// Bytes consumed past this are dropped from the front of the buffer.
29const COMPACT_PAST: usize = 1 << 16;
30
31/// A recording decoded as its bytes come.
32#[derive(Default)]
33pub struct Stream {
34    buf: Vec<u8>,
35    at: usize,
36    header: bool,
37    dec: Decoder,
38    trace: Trace,
39    timeline: Timeline,
40}
41
42impl Stream {
43    pub fn new() -> Self {
44        Stream {
45            trace: Trace::new(),
46            ..Default::default()
47        }
48    }
49
50    /// Decodes what `bytes` complete: how many records came. A cut record
51    /// waits for the next push; bytes that are no recording are an error,
52    /// and so is a header of another version.
53    pub fn push(&mut self, bytes: &[u8]) -> Result<usize, Error> {
54        self.buf.extend_from_slice(bytes);
55        if !self.header {
56            match check_header(&self.buf)? {
57                Some(n) => {
58                    self.at = n;
59                    self.header = true;
60                }
61                None => return Ok(0),
62            }
63        }
64        let from = self.trace.records.len();
65        let mut came = 0;
66        while let Some((tag, body)) = framed(&self.buf, self.at) {
67            self.at = body.end;
68            // A body shorter than its fields is skipped, as a reader of a
69            // whole file skips it.
70            if let Some(Some(r)) = self.dec.decode(tag, &self.buf[body]) {
71                self.trace.push(r);
72                came += 1;
73            }
74        }
75        self.timeline.extend(&self.trace.records[from..]);
76        if self.at > COMPACT_PAST {
77            self.buf.drain(..self.at);
78            self.at = 0;
79        }
80        Ok(came)
81    }
82
83    /// Starts again, as a source that began a new recording asks.
84    pub fn reset(&mut self) {
85        *self = Stream::new();
86    }
87
88    /// The recording so far.
89    pub fn trace(&self) -> &Trace {
90        &self.trace
91    }
92
93    /// The timeline so far.
94    pub fn timeline(&self) -> &Timeline {
95        &self.timeline
96    }
97
98    /// Whether the header has come.
99    pub fn started(&self) -> bool {
100        self.header
101    }
102
103    /// The bytes of a record cut short, waiting for the rest.
104    pub fn pending(&self) -> usize {
105        self.buf.len() - self.at
106    }
107
108    /// Reads what `t` has into the stream: [`Polled::More`] with the
109    /// records that came, [`Polled::Idle`], [`Polled::Ended`], or
110    /// [`Polled::Reset`] after starting again for a new recording.
111    pub fn poll(&mut self, t: &mut dyn Transport) -> Result<Polled, PollError> {
112        let mut buf = Vec::new();
113        match t.poll(&mut buf).map_err(PollError::Io)? {
114            Polled::Reset => {
115                self.reset();
116                self.push(&buf).map_err(PollError::Trace)?;
117                Ok(Polled::Reset)
118            }
119            Polled::More(_) => Ok(Polled::More(self.push(&buf).map_err(PollError::Trace)?)),
120            other => Ok(other),
121        }
122    }
123}
124
125/// What a [`Transport`] had.
126#[derive(Clone, Copy, Debug, PartialEq, Eq)]
127pub enum Polled {
128    /// Bytes, appended to the buffer: how many (from [`Stream::poll`], how
129    /// many records they completed).
130    More(usize),
131    /// Nothing yet; more may come.
132    Idle,
133    /// Nothing more will come.
134    Ended,
135    /// The source began again — a file rewritten, a run restarted: what
136    /// was read is another recording's, and the buffer holds the new one
137    /// from its start.
138    Reset,
139}
140
141/// Where a recording's bytes come from (ADR 0075, decision 5): a file
142/// ([`FileTransport`]), and later a pipe or a socket, behind one door.
143pub trait Transport: Send {
144    /// Appends what came since the last poll to `buf`, without waiting.
145    fn poll(&mut self, buf: &mut Vec<u8>) -> io::Result<Polled>;
146}
147
148/// Why [`Stream::poll`] stopped.
149#[derive(Debug)]
150pub enum PollError {
151    Io(io::Error),
152    Trace(Error),
153}
154
155impl std::fmt::Display for PollError {
156    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
157        match self {
158            PollError::Io(e) => write!(f, "{e}"),
159            PollError::Trace(e) => write!(f, "{e}"),
160        }
161    }
162}
163
164impl std::error::Error for PollError {}
kui_trace/stream.rs

kui_trace/
stream.rs
