fast_pull/pusher/
mem.rs

1//! In-memory pusher backed by a shared `Vec<u8>`.
2
3use crate::{ProgressEntry, ProgressListener, Pusher};
4use bytes::Bytes;
5use parking_lot::Mutex;
6use std::{sync::Arc, vec::Vec};
7
8/// In-memory pusher for testing or buffer-based workflows.
9///
10/// All pushed data is stored in a shared `Vec<u8>` protected by a mutex.
11/// Supports random-access writes via `copy_from_slice` for non-sequential ranges.
12#[derive(Default)]
13pub struct MemPusher {
14    /// The shared buffer the pusher writes into.
15    pub receive: Arc<Mutex<Vec<u8>>>,
16    listener: Option<ProgressListener>,
17}
18impl Clone for MemPusher {
19    fn clone(&self) -> Self {
20        Self {
21            receive: self.receive.clone(),
22            listener: None,
23        }
24    }
25}
26impl MemPusher {
27    /// Create an empty in-memory pusher.
28    #[must_use]
29    pub fn new() -> Self {
30        Self {
31            receive: Arc::new(Mutex::new(Vec::new())),
32            listener: None,
33        }
34    }
35    /// Create an in-memory pusher with a pre-allocated buffer of `capacity` bytes.
36    #[must_use]
37    pub fn with_capacity(capacity: usize) -> Self {
38        Self {
39            receive: Arc::new(Mutex::new(Vec::with_capacity(capacity))),
40            listener: None,
41        }
42    }
43}
44impl std::fmt::Debug for MemPusher {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("MemPusher")
47            .field("receive", &self.receive)
48            .finish_non_exhaustive()
49    }
50}
51impl Pusher for MemPusher {
52    type Error = std::convert::Infallible;
53
54    fn set_listener(&mut self, cb: ProgressListener) {
55        self.listener = Some(cb);
56    }
57
58    fn push(&mut self, range: &ProgressEntry, content: Bytes) -> Result<(), (Self::Error, Bytes)> {
59        #![allow(clippy::significant_drop_tightening, clippy::cast_possible_truncation)]
60        let mut guard = self.receive.lock();
61        if range.start as usize == guard.len() {
62            guard.extend_from_slice(&content);
63        } else {
64            if guard.len() < range.end as usize {
65                guard.resize(range.end as usize, 0);
66            }
67            guard[range.start as usize..range.end as usize].copy_from_slice(&content);
68        }
69        drop(guard);
70        if let Some(l) = &mut self.listener {
71            l(range.clone());
72        }
73        Ok(())
74    }
75}
76
77#[cfg(test)]
78mod tests {
79    #![allow(clippy::unwrap_used)]
80    use super::*;
81    use std::sync::{Arc, Mutex};
82
83    #[test]
84    fn sequential_append() {
85        let mut p = MemPusher::new();
86        p.push(&(0..3), Bytes::copy_from_slice(b"abc")).unwrap();
87        p.push(&(3..6), Bytes::copy_from_slice(b"def")).unwrap();
88        assert_eq!(&p.receive.lock()[..], b"abcdef");
89    }
90
91    #[test]
92    fn random_access_resizes_and_writes() {
93        let mut p = MemPusher::new();
94        p.push(&(5..8), Bytes::copy_from_slice(b"xyz")).unwrap();
95        // The gap before the write is filled with zeros via resize.
96        assert_eq!(p.receive.lock().len(), 8);
97        p.push(&(0..5), Bytes::copy_from_slice(b"hello")).unwrap();
98        assert_eq!(&p.receive.lock()[..], b"helloxyz");
99    }
100
101    #[test]
102    fn empty_push_is_noop() {
103        let mut p = MemPusher::new();
104        p.push(&(0..0), Bytes::new()).unwrap();
105        assert!(p.receive.lock().is_empty());
106    }
107
108    #[test]
109    fn listener_invoked_with_range() {
110        let mut p = MemPusher::new();
111        let seen = Arc::new(Mutex::new(None::<ProgressEntry>));
112        let seen2 = seen.clone();
113        p.set_listener(Box::new(move |r| {
114            *seen2.lock().unwrap() = Some(r);
115        }));
116        p.push(&(0..4), Bytes::copy_from_slice(b"data")).unwrap();
117        assert_eq!(*seen.lock().unwrap(), Some(0..4));
118    }
119
120    #[test]
121    fn clone_shares_receive() {
122        let mut p = MemPusher::new();
123        p.push(&(0..2), Bytes::copy_from_slice(b"hi")).unwrap();
124        let mut q = p.clone();
125        q.push(&(2..4), Bytes::copy_from_slice(b"ya")).unwrap();
126        // Both handles observe the same underlying vec.
127        assert_eq!(&p.receive.lock()[..], b"hiya");
128    }
129
130    #[test]
131    fn debug_impl() {
132        // Lines 42-46: the `Debug` impl for `MemPusher`.
133        let p = MemPusher::new();
134        let _ = format!("{p:?}");
135    }
136}
fast_pull/pusher/mem.rs

fast_pull/pusher/
mem.rs
