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

fast_pull/pusher/
mem.rs
