epochfs 0.1.5

EpochFS is a versioned cloud file system with git-like branching, transaction support.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

use anyhow::Result;
use futures::{Stream, StreamExt, TryStreamExt};
use opendal::Buffer;
use std::mem;
use std::pin::pin;

use crate::Fs;

const DEFAULT_CHUNK_SIZE: usize = 8 * 1024 * 1024;

/// File represents a file in the epochfs.
#[derive(Debug, Clone)]
pub struct File {
    fs: Fs,

    pub(crate) path: String,
    pub(crate) chunks: Vec<String>,
}

impl File {
    /// Create a new file.
    pub(crate) fn new(fs: Fs, path: String) -> Self {
        Self::with_chunks(fs, path, Vec::new())
    }

    /// Create a new file.
    pub(crate) fn with_chunks(fs: Fs, path: String, chunks: Vec<String>) -> Self {
        Self { fs, path, chunks }
    }

    /// Get the path and chunks of the file.
    pub(crate) fn into_parts(self) -> (String, Vec<String>) {
        (self.path, self.chunks)
    }

    /// Get the path of the file.
    pub fn path(&self) -> &str {
        &self.path
    }

    /// Write given buffer to the file.
    ///
    /// This function will calculate the chunk id from the buffer and write
    /// the buffer to the storage. If the chunk id already exists, we can
    /// reuse the existing chunk instead of creating a new one.
    ///
    /// # TODO
    ///
    /// We can use if-not-exists to save an extra request.
    pub async fn write(&mut self, bs: Buffer) -> Result<()> {
        let chunk_id = self.fs.write_chunk(bs).await?;
        self.chunks.push(chunk_id);
        Ok(())
    }

    /// Write a stream of buffers to the file.
    ///
    /// sink will make sure that all chunks are aligned with 8MiB.
    pub async fn sink(&mut self, mut stream: impl Stream<Item = Result<Buffer>>) -> Result<()> {
        let mut stream = pin!(stream);

        let mut chunks = Vec::new();
        let mut size = 0;
        while let Some(bs) = stream.next().await {
            let bs = bs?;
            if size + bs.len() < DEFAULT_CHUNK_SIZE {
                size += bs.len();
                chunks.push(bs);
                continue;
            }

            let consume_size = DEFAULT_CHUNK_SIZE - size;
            // Push the last chunk.
            chunks.push(bs.slice(0..consume_size));
            self.write(Buffer::from_iter(
                mem::take(&mut chunks).into_iter().flatten(),
            ))
            .await?;

            if consume_size < bs.len() {
                chunks.push(bs.slice(consume_size..));
                size += bs.len() - consume_size;
            } else {
                size = 0
            }
        }
        if size > 0 {
            self.write(Buffer::from_iter(
                mem::take(&mut chunks).into_iter().flatten(),
            ))
            .await?;
        }
        Ok(())
    }

    /// Commit the file to the database.
    pub async fn commit(&mut self) -> Result<()> {
        self.fs.commit_file(&self.path, self.chunks.clone()).await?;
        Ok(())
    }

    /// Read given file into buffer.
    pub async fn read(&self) -> Result<Buffer> {
        let buffers: Vec<_> = self.stream().await?.try_collect().await?;
        Ok(Buffer::from_iter(buffers.into_iter().flatten()))
    }

    /// Stream the entire content file in buffers.
    pub async fn stream(&self) -> Result<impl Stream<Item = Result<Buffer>>> {
        let fs = self.fs.clone();
        let stream = futures::stream::iter(self.chunks.clone()).then(move |v| {
            let fs = fs.clone();
            async move { fs.read_chunk(&v).await }
        });
        Ok(stream)
    }
}