fast_pull/pusher/
cache_std.rs

1//! High-level file pusher combining caching, buffering, and standard file I/O.
2
3use crate::{
4    BufWriterPusher, CacheSeqPusher, ProgressEntry, ProgressListener, Pusher, StdFilePusher,
5};
6use bytes::Bytes;
7
8/// File pusher combining [`CacheSeqPusher`] with [`StdFilePusher`].
9///
10/// Provides out-of-order reordering on top of standard file I/O.
11/// The write buffer, watermark levels, and sync-all behavior are forwarded
12/// to [`StdFilePusher::new`].
13#[derive(Debug)]
14pub struct CacheFilePusher {
15    inner: CacheSeqPusher<BufWriterPusher<StdFilePusher>>,
16}
17
18impl CacheFilePusher {
19    /// # Errors
20    /// 1. Returns an error if `fs::set_len` fails.
21    /// 2. Returns an error if [`StdFilePusher`] initialization fails.
22    pub async fn new(
23        file: tokio::fs::File,
24        size: u64,
25        sync_all: bool,
26        high_watermark: usize,
27        low_watermark: usize,
28        buffer_size: usize,
29    ) -> std::io::Result<Self> {
30        let file_pusher =
31            BufWriterPusher::new(StdFilePusher::new(file, size, sync_all).await?, buffer_size);
32        let inner = CacheSeqPusher::new(file_pusher, high_watermark, low_watermark);
33        Ok(Self { inner })
34    }
35}
36
37impl Pusher for CacheFilePusher {
38    type Error = std::io::Error;
39
40    fn set_listener(&mut self, cb: ProgressListener) {
41        self.inner.set_listener(cb);
42    }
43
44    #[inline]
45    fn push(&mut self, range: &ProgressEntry, bytes: Bytes) -> Result<(), (Self::Error, Bytes)> {
46        self.inner.push(range, bytes)
47    }
48
49    #[inline]
50    fn flush(&mut self) -> Result<(), Self::Error> {
51        self.inner.flush()
52    }
53}
54
55#[cfg(test)]
56mod tests {
57    #![allow(clippy::unwrap_used)]
58    use super::*;
59    use std::{fs::File, io::Read, vec::Vec};
60    use tempfile::NamedTempFile;
61
62    #[tokio::test]
63    async fn test_cache_file_pusher() {
64        let temp_file = NamedTempFile::new().unwrap();
65        let file_path = temp_file.path();
66
67        // Initialize the wrapper:
68        // cache size = 1MB, watermark = 512KB, BufWriter = 8KB
69        let mut pusher = CacheFilePusher::new(
70            temp_file.reopen().unwrap().into(),
71            10,
72            false,
73            1024 * 1024,
74            512 * 1024,
75            8 * 1024,
76        )
77        .await
78        .unwrap();
79
80        // Write data
81        let data = b"234";
82        let range = 2..5;
83        pusher.push(&range, data[..].into()).unwrap();
84        pusher.flush().unwrap();
85
86        // Verify file content
87        let mut file_content = Vec::new();
88        File::open(file_path)
89            .unwrap()
90            .read_to_end(&mut file_content)
91            .unwrap();
92        assert_eq!(file_content, b"\0\x00234\0\0\0\0\0");
93    }
94}
fast_pull/pusher/cache_std.rs

fast_pull/pusher/
cache_std.rs
