19946635754

Committed 04 Dec 2025 10:56PM UTC coverage: 58.306% (+0.5%) from 57.787%

Build # 19946635754

Build Type

Pull #1078

github

Committed by

web-flow

Commit Message

Merge 286a539ef into cb87d63aa

Pull Request Pull Request #1078: feat(facet-json): streaming JSON deserializer

Run Details

1253 of 1541 new or added lines in 6 files covered. (81.31%)

8 existing lines in 2 files now uncovered.

23561 of 40409 relevant lines covered (58.31%)

499.03 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.65

/facet-json/src/scan_buffer.rs

//! Buffer management for streaming JSON scanning.
//!
//! # Design: No Compaction Needed
//!
//! This buffer uses a grow-only strategy that never requires compacting
//! (shifting data to the left). This is possible because:
//!
//! 1. When the scanner returns a complete token, we materialize it immediately.
//!    The raw bytes are no longer needed.
//!
//! 2. When the scanner returns `Eof` (end of buffer, not real EOF), all tokens
//!    in the buffer have been processed. We can **reset** and read fresh data.
//!
//! 3. When the scanner returns `NeedMore` (mid-token), we **grow** the buffer
//!    and read more data into the new space. Indices remain valid.
//!
//! This avoids all data copying, which is both simpler and more efficient.

use alloc::vec::Vec;

/// Default buffer capacity (8KB)
pub const DEFAULT_CAPACITY: usize = 8 * 1024;

/// A refillable buffer for streaming JSON parsing.
///
/// Uses a grow-only strategy: we either reset (when all data processed)
/// or grow (when mid-token). Never compacts/shifts data.
#[derive(Debug)]
pub struct ScanBuffer {
    /// The underlying buffer
    data: Vec<u8>,
    /// How many bytes are valid (filled with data)
    filled: usize,
    /// Whether EOF has been reached on the underlying reader
    eof: bool,
}

impl ScanBuffer {
    /// Create a new buffer with the default capacity.
    pub fn new() -> Self {
        Self::with_capacity(DEFAULT_CAPACITY)
    }

    /// Create a new buffer with a specific capacity.
    pub fn with_capacity(capacity: usize) -> Self {
        Self {
            data: vec![0u8; capacity],
            filled: 0,
            eof: false,
        }
    }

    /// Create a buffer from an existing slice (for slice-based parsing).
    pub fn from_slice(slice: &[u8]) -> Self {
        let mut data = Vec::with_capacity(slice.len());
        data.extend_from_slice(slice);
        Self {
            filled: data.len(),
            data,
            eof: true, // No more data to read
        }
    }

    /// Get the current buffer contents.
    #[inline]
    pub fn data(&self) -> &[u8] {
        &self.data[..self.filled]
    }

    /// Whether the underlying reader has reached EOF.
    #[inline]
    pub fn is_eof(&self) -> bool {
        self.eof
    }

    /// How many bytes are filled.
    #[inline]
    pub fn filled(&self) -> usize {
        self.filled
    }

    /// Get the buffer's total capacity.
    #[inline]
    pub fn capacity(&self) -> usize {
        self.data.len()
    }

    /// Reset the buffer for fresh data.
    ///
    /// Called when all data has been processed (scanner returned Eof but reader has more).
    /// This is NOT compaction - we're simply starting fresh because everything was consumed.
    pub fn reset(&mut self) {
        self.filled = 0;
        // Note: we don't reset eof here - that's determined by the reader
    }

    /// Grow the buffer to make room for more data.
    ///
    /// Called when mid-token (NeedMore) and buffer is full.
    /// We grow rather than compact because:
    /// - No data copying needed
    /// - Scanner indices remain valid
    /// - Simpler logic
    pub fn grow(&mut self) {
        let new_capacity = self.data.len() * 2;
        self.data.resize(new_capacity, 0);
    }

    /// Refill the buffer from a synchronous reader.
    ///
    /// Reads more data into the unfilled portion of the buffer.
    /// Returns the number of bytes read, or 0 if EOF.
    #[cfg(feature = "std")]
    pub fn refill<R: std::io::Read>(&mut self, reader: &mut R) -> std::io::Result<usize> {
        if self.eof {
            return Ok(0);
        }

        let read_buf = &mut self.data[self.filled..];
        if read_buf.is_empty() {
            // Buffer is full - caller should grow() first if needed
            return Ok(0);
        }

        let n = reader.read(read_buf)?;
        self.filled += n;

        if n == 0 {
            self.eof = true;
        }

        Ok(n)
    }

    /// Refill the buffer from an async reader (tokio).
    #[cfg(feature = "tokio")]
    pub async fn refill_tokio<R>(&mut self, reader: &mut R) -> std::io::Result<usize>
    where
        R: tokio::io::AsyncRead + Unpin,
    {
        use tokio::io::AsyncReadExt;

        if self.eof {
            return Ok(0);
        }

        let read_buf = &mut self.data[self.filled..];
        if read_buf.is_empty() {
            return Ok(0);
        }

        let n = reader.read(read_buf).await?;
        self.filled += n;

        if n == 0 {
            self.eof = true;
        }

        Ok(n)
    }

    /// Refill the buffer from an async reader (futures-io).
    #[cfg(feature = "futures-io")]
    pub async fn refill_futures<R>(&mut self, reader: &mut R) -> std::io::Result<usize>
    where
        R: futures_io::AsyncRead + Unpin,
    {
        use core::pin::Pin;
        use core::task::Context;

        if self.eof {
            return Ok(0);
        }

        let read_buf = &mut self.data[self.filled..];
        if read_buf.is_empty() {
            return Ok(0);
        }

        let n = core::future::poll_fn(|cx: &mut Context<'_>| {
            Pin::new(&mut *reader).poll_read(cx, read_buf)
        })
        .await?;
        self.filled += n;

        if n == 0 {
            self.eof = true;
        }

        Ok(n)
    }
}

impl Default for ScanBuffer {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_from_slice() {
        let buf = ScanBuffer::from_slice(b"hello world");
        assert_eq!(buf.data(), b"hello world");
        assert_eq!(buf.filled(), 11);
        assert!(buf.is_eof());
    }

    #[test]
    fn test_reset() {
        let mut buf = ScanBuffer::from_slice(b"hello");
        assert_eq!(buf.filled(), 5);
        buf.reset();
        assert_eq!(buf.filled(), 0);
        assert_eq!(buf.data(), b"");
    }

    #[test]
    fn test_grow() {
        let mut buf = ScanBuffer::with_capacity(4);
        assert_eq!(buf.capacity(), 4);
        buf.grow();
        assert_eq!(buf.capacity(), 8);
    }

    #[cfg(feature = "std")]
    #[test]
    fn test_refill_from_reader() {
        use std::io::Cursor;

        let mut reader = Cursor::new(b"hello world");
        let mut buf = ScanBuffer::with_capacity(8);

        // First read
        let n = buf.refill(&mut reader).unwrap();
        assert_eq!(n, 8);
        assert_eq!(buf.data(), b"hello wo");

        // Buffer full, refill returns 0
        let n = buf.refill(&mut reader).unwrap();
        assert_eq!(n, 0);

        // Grow and refill
        buf.grow();
        let n = buf.refill(&mut reader).unwrap();
        assert_eq!(n, 3);
        assert_eq!(buf.data(), b"hello world");
        assert!(!buf.is_eof()); // Not EOF yet - we got 3 bytes

        // One more refill to confirm EOF
        let n = buf.refill(&mut reader).unwrap();
        assert_eq!(n, 0);
        assert!(buf.is_eof());
    }
}

1	//! Buffer management for streaming JSON scanning.
2	//!
3	//! # Design: No Compaction Needed
4	//!
5	//! This buffer uses a grow-only strategy that never requires compacting
6	//! (shifting data to the left). This is possible because:
7	//!
8	//! 1. When the scanner returns a complete token, we materialize it immediately.
9	//! The raw bytes are no longer needed.
10	//!
11	//! 2. When the scanner returns `Eof` (end of buffer, not real EOF), all tokens
12	//! in the buffer have been processed. We can reset and read fresh data.
13	//!
14	//! 3. When the scanner returns `NeedMore` (mid-token), we grow the buffer
15	//! and read more data into the new space. Indices remain valid.
16	//!
17	//! This avoids all data copying, which is both simpler and more efficient.
18
19	use alloc::vec::Vec;
20
21	/// Default buffer capacity (8KB)
22	pub const DEFAULT_CAPACITY: usize = 8 * 1024;
23
24	/// A refillable buffer for streaming JSON parsing.
25	///
26	/// Uses a grow-only strategy: we either reset (when all data processed)
27	/// or grow (when mid-token). Never compacts/shifts data.
28	#[derive(Debug)]
29	pub struct ScanBuffer {
30	/// The underlying buffer
31	data: Vec<u8>,
32	/// How many bytes are valid (filled with data)
33	filled: usize,
34	/// Whether EOF has been reached on the underlying reader
35	eof: bool,
36	}
37
38	impl ScanBuffer {
39	/// Create a new buffer with the default capacity.
40	pub fn new() -> Self {	3✔
41	Self::with_capacity(DEFAULT_CAPACITY)	3✔
42	}	3✔
43
44	/// Create a new buffer with a specific capacity.
45	pub fn with_capacity(capacity: usize) -> Self {	9✔
46	Self {	9✔
47	data: vec![0u8; capacity],	9✔
48	filled: 0,	9✔
49	eof: false,	9✔
50	}	9✔
51	}	9✔
52
53	/// Create a buffer from an existing slice (for slice-based parsing).
54	pub fn from_slice(slice: &[u8]) -> Self {	2✔
55	let mut data = Vec::with_capacity(slice.len());	2✔
56	data.extend_from_slice(slice);	2✔
57	Self {	2✔
58	filled: data.len(),	2✔
59	data,	2✔
60	eof: true, // No more data to read	2✔
61	}	2✔
62	}	2✔
63
64	/// Get the current buffer contents.
65	#[inline]
66	pub fn data(&self) -> &[u8] {	126✔
67	&self.data[..self.filled]	126✔
68	}	126✔
69
70	/// Whether the underlying reader has reached EOF.
71	#[inline]
72	pub fn is_eof(&self) -> bool {	26✔
73	self.eof	26✔
74	}	26✔
75
76	/// How many bytes are filled.
77	#[inline]
78	pub fn filled(&self) -> usize {	96✔
79	self.filled	96✔
80	}	96✔
81
82	/// Get the buffer's total capacity.
83	#[inline]
84	pub fn capacity(&self) -> usize {	18✔
85	self.data.len()	18✔
86	}	18✔
87
88	/// Reset the buffer for fresh data.
89	///
90	/// Called when all data has been processed (scanner returned Eof but reader has more).
91	/// This is NOT compaction - we're simply starting fresh because everything was consumed.
92	pub fn reset(&mut self) {	17✔
93	self.filled = 0;	17✔
94	// Note: we don't reset eof here - that's determined by the reader
95	}	17✔
96
97	/// Grow the buffer to make room for more data.
98	///
99	/// Called when mid-token (NeedMore) and buffer is full.
100	/// We grow rather than compact because:
101	/// - No data copying needed
102	/// - Scanner indices remain valid
103	/// - Simpler logic
104	pub fn grow(&mut self) {	7✔
105	let new_capacity = self.data.len() * 2;	7✔
106	self.data.resize(new_capacity, 0);	7✔
107	}	7✔
108
109	/// Refill the buffer from a synchronous reader.
110	///
111	/// Reads more data into the unfilled portion of the buffer.
112	/// Returns the number of bytes read, or 0 if EOF.
113	#[cfg(feature = "std")]
114	pub fn refill<R: std::io::Read>(&mut self, reader: &mut R) -> std::io::Result<usize> {	43✔
115	if self.eof {	43✔
NEW 116	return Ok(0);	×
117	}	43✔
118
119	let read_buf = &mut self.data[self.filled..];	43✔
120	if read_buf.is_empty() {	43✔
121	// Buffer is full - caller should grow() first if needed
122	return Ok(0);	1✔
123	}	42✔
124
125	let n = reader.read(read_buf)?;	42✔
126	self.filled += n;	42✔
127
128	if n == 0 {	42✔
129	self.eof = true;	8✔
130	}	34✔
131
132	Ok(n)	42✔
133	}	43✔
134
135	/// Refill the buffer from an async reader (tokio).
136	#[cfg(feature = "tokio")]
137	pub async fn refill_tokio<R>(&mut self, reader: &mut R) -> std::io::Result<usize>
138	where
139	R: tokio::io::AsyncRead + Unpin,
140	{
141	use tokio::io::AsyncReadExt;
142
143	if self.eof {
144	return Ok(0);
145	}
146
147	let read_buf = &mut self.data[self.filled..];
148	if read_buf.is_empty() {
149	return Ok(0);
150	}
151
152	let n = reader.read(read_buf).await?;
153	self.filled += n;
154
155	if n == 0 {
156	self.eof = true;
157	}
158
159	Ok(n)
160	}
161
162	/// Refill the buffer from an async reader (futures-io).
163	#[cfg(feature = "futures-io")]
164	pub async fn refill_futures<R>(&mut self, reader: &mut R) -> std::io::Result<usize>
165	where
166	R: futures_io::AsyncRead + Unpin,
167	{
168	use core::pin::Pin;
169	use core::task::Context;
170
171	if self.eof {
172	return Ok(0);
173	}
174
175	let read_buf = &mut self.data[self.filled..];
176	if read_buf.is_empty() {
177	return Ok(0);
178	}
179
180	let n = core::future::poll_fn(\|cx: &mut Context<'_>\| {
181	Pin::new(&mut *reader).poll_read(cx, read_buf)
182	})
183	.await?;
184	self.filled += n;
185
186	if n == 0 {
187	self.eof = true;
188	}
189
190	Ok(n)
191	}
192	}
193
194	impl Default for ScanBuffer {
NEW 195	fn default() -> Self {	×
NEW 196	Self::new()	×
NEW 197	}	×
198	}
199
200	#[cfg(test)]
201	mod tests {
202	use super::*;
203
204	#[test]
205	fn test_from_slice() {	1✔
206	let buf = ScanBuffer::from_slice(b"hello world");	1✔
207	assert_eq!(buf.data(), b"hello world");	1✔
208	assert_eq!(buf.filled(), 11);	1✔
209	assert!(buf.is_eof());	1✔
210	}	1✔
211
212	#[test]
213	fn test_reset() {	1✔
214	let mut buf = ScanBuffer::from_slice(b"hello");	1✔
215	assert_eq!(buf.filled(), 5);	1✔
216	buf.reset();	1✔
217	assert_eq!(buf.filled(), 0);	1✔
218	assert_eq!(buf.data(), b"");	1✔
219	}	1✔
220
221	#[test]
222	fn test_grow() {	1✔
223	let mut buf = ScanBuffer::with_capacity(4);	1✔
224	assert_eq!(buf.capacity(), 4);	1✔
225	buf.grow();	1✔
226	assert_eq!(buf.capacity(), 8);	1✔
227	}	1✔
228
229	#[cfg(feature = "std")]
230	#[test]
231	fn test_refill_from_reader() {	1✔
232	use std::io::Cursor;
233
234	let mut reader = Cursor::new(b"hello world");	1✔
235	let mut buf = ScanBuffer::with_capacity(8);	1✔
236
237	// First read
238	let n = buf.refill(&mut reader).unwrap();	1✔
239	assert_eq!(n, 8);	1✔
240	assert_eq!(buf.data(), b"hello wo");	1✔
241
242	// Buffer full, refill returns 0
243	let n = buf.refill(&mut reader).unwrap();	1✔
244	assert_eq!(n, 0);	1✔
245
246	// Grow and refill
247	buf.grow();	1✔
248	let n = buf.refill(&mut reader).unwrap();	1✔
249	assert_eq!(n, 3);	1✔
250	assert_eq!(buf.data(), b"hello world");	1✔
251	assert!(!buf.is_eof()); // Not EOF yet - we got 3 bytes	1✔
252
253	// One more refill to confirm EOF
254	let n = buf.refill(&mut reader).unwrap();	1✔
255	assert_eq!(n, 0);	1✔
256	assert!(buf.is_eof());	1✔
257	}	1✔
258	}

facet-rs / facet / 19946635754

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous