spray 0.1.0

spray
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

use crate::files;
use crate::files::*;
use crate::types::Result;
use std::io::SeekFrom;
use tokio::io::AsyncWrite;

pub struct Log {
    pub id: u64,
    pub position: u64,
    file: File,
}

impl Log {
    pub async fn new(location: &PathBuf, id: u64) -> Result<Log> {
        let path = location.join(id.to_string()).with_extension("log");
        let file = files::create(&path)?;
        let position = 0;
        Ok(Log { id, file, position })
    }
    pub async fn open(location: &PathBuf, id: u64) -> Result<Log> {
        let path = location.join(id.to_string()).with_extension("log");
        let file = files::open(&path)?;
        let position = file.metadata().await.unwrap().len();
        Ok(Log { id, file, position })
    }

    pub async fn add_entry(&mut self, offset: u64, entry: Vec<u8>) -> Result<()> {
        let entry_size = entry.len() as u32;
        let total_size = 8 + 4 + entry_size as usize; // offset + entry size + entry
        let mut bytes = Vec::with_capacity(total_size);
        bytes.extend(offset.to_be_bytes().iter());
        bytes.extend(entry_size.to_be_bytes().iter());
        bytes.extend(entry.iter());
        self.file.write_all(&bytes).await?;
        self.file.flush().await?;
        self.position += total_size as u64;
        Ok(())
    }

    //TODO: use zerocopy optimisation using sendfile
    pub async fn stream_entries<T>(
        &mut self,
        position: u64,
        bytes: usize,
        target: &mut T,
    ) -> Result<()>
    where
        T: AsyncWrite + Unpin + ?Sized,
    {
        let max_bytes = self.size().await as usize;
        let bytes = if bytes > max_bytes { max_bytes } else { bytes };
        self.file.seek(SeekFrom::Start(position)).await?;
        let mut buf = vec![0; bytes];
        self.file.read_exact(&mut buf).await?;
        target.write(&buf).await?;
        self.file.seek(SeekFrom::End(0)).await?;
        Ok(())
    }

    pub async fn size(&self) -> u64 {
        self.file.metadata().await.unwrap().len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_util as test;

    #[tokio::test]
    async fn create_new_log() {
        let location = test::create_a_test_directory();

        let log = Log::new(&location, 123).await.unwrap();

        assert_eq!(log.position, 0);
        assert_eq!(log.size().await, 0);
    }

    #[tokio::test]
    async fn add_entry() {
        let id = 5000;
        let location = test::create_a_test_directory();
        let mut log = Log::new(&location, id).await.unwrap();

        log.add_entry(5000, vec![1, 2, 3]).await.unwrap();
        log.add_entry(5001, vec![3, 4]).await.unwrap();

        assert_eq!(log.position, 29);
    }

    #[tokio::test]
    async fn open_existing_log() {
        let location = test::create_a_test_directory();
        let id = 10;
        let mut log = Log::new(&location, id).await.unwrap();
        log.add_entry(5000, vec![1, 2, 3]).await.unwrap();

        let log = Log::open(&location, id).await.unwrap();
        assert_eq!(log.position, 15);
    }

    #[tokio::test]
    async fn stream_entries() {
        let id = 10;
        let location = test::create_a_test_directory();
        let mut log = Log::new(&location, id).await.unwrap();
        log.add_entry(10, vec![1, 2]).await.unwrap();
        log.add_entry(11, vec![3, 4]).await.unwrap();

        let mut entries = Vec::new();
        log.stream_entries(0, 16000, &mut entries).await.unwrap();

        assert_eq!(
            &entries,
            &[
                0, 0, 0, 0, 0, 0, 0, 10, //offset
                0, 0, 0, 2, //len
                1, 2, //entry
                0, 0, 0, 0, 0, 0, 0, 11, //offset
                0, 0, 0, 2, //len
                3, 4 //entry
            ]
        );
    }

    #[tokio::test]
    async fn stream_partial() {
        let id = 10;
        let location = test::create_a_test_directory();
        let mut log = Log::new(&location, id).await.unwrap();
        log.add_entry(10, vec![1, 2]).await.unwrap();
        log.add_entry(11, vec![3, 4]).await.unwrap();

        let mut entries = Vec::new();
        log.stream_entries(0, 14, &mut entries).await.unwrap();

        assert_eq!(
            &entries,
            &[
                0, 0, 0, 0, 0, 0, 0, 10, //offset
                0, 0, 0, 2, //len
                1, 2, //entry
            ]
        );
    }
}