vortex_io/std_file/
write.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::fs::File;
5use std::future::Future;
6use std::io;
7use std::io::Write;
8use std::path::Path;
9use std::sync::Arc;
10
11use parking_lot::Mutex;
12
13use crate::IoBuf;
14use crate::VortexWrite;
15use crate::runtime::Handle;
16
17/// A local file writer that runs blocking filesystem operations on a Vortex runtime.
18pub struct FileWrite {
19    file: Arc<Mutex<File>>,
20    handle: Handle,
21}
22
23impl FileWrite {
24    /// Create or truncate a file using `handle` for the blocking open operation.
25    pub async fn create(path: impl AsRef<Path>, handle: Handle) -> io::Result<Self> {
26        let path = path.as_ref().to_path_buf();
27        let file = handle.spawn_blocking(move || File::create(path)).await?;
28        Ok(Self {
29            file: Arc::new(Mutex::new(file)),
30            handle,
31        })
32    }
33}
34
35impl VortexWrite for FileWrite {
36    fn write_all<B: IoBuf>(&mut self, buffer: B) -> impl Future<Output = io::Result<B>> + Send {
37        let file = Arc::clone(&self.file);
38        let handle = self.handle.clone();
39        async move {
40            handle
41                .spawn_blocking(move || {
42                    file.lock().write_all(buffer.as_slice())?;
43                    Ok(buffer)
44                })
45                .await
46        }
47    }
48
49    fn flush(&mut self) -> impl Future<Output = io::Result<()>> + Send {
50        let file = Arc::clone(&self.file);
51        let handle = self.handle.clone();
52        async move { handle.spawn_blocking(move || file.lock().flush()).await }
53    }
54
55    fn shutdown(&mut self) -> impl Future<Output = io::Result<()>> + Send {
56        self.flush()
57    }
58}
59
60#[cfg(test)]
61mod tests {
62    use tempfile::tempdir;
63    use vortex_error::VortexResult;
64
65    use super::*;
66    use crate::runtime::BlockingRuntime;
67    use crate::runtime::current::CurrentThreadRuntime;
68
69    #[test]
70    fn test_write_file() -> VortexResult<()> {
71        let directory = tempdir()?;
72        let path = directory.path().join("write.vortex");
73        let runtime = CurrentThreadRuntime::new();
74        runtime.block_on(async {
75            let mut writer = FileWrite::create(&path, runtime.handle()).await?;
76            writer.write_all(vec![1, 2, 3]).await?;
77            writer.shutdown().await
78        })?;
79
80        assert_eq!(std::fs::read(path)?, vec![1, 2, 3]);
81        Ok(())
82    }
83}
vortex_io/std_file/write.rs

vortex_io/std_file/
write.rs
