gix_pack/data/input/
bytes_to_entries.rs

1use std::{fs, io};
2
3use gix_hash::{Hasher, ObjectId};
4use gix_zlib::Decompress;
5
6use crate::data::input;
7
8/// An iterator over [`Entries`][input::Entry] in a byte stream.
9///
10/// The iterator used as part of [`Bundle::write_to_directory(…)`][crate::Bundle::write_to_directory()].
11pub struct BytesToEntriesIter<BR> {
12    read: BR,
13    decompressor: Decompress,
14    offset: u64,
15    had_error: bool,
16    version: crate::data::Version,
17    objects_left: u32,
18    hash: Option<Hasher>,
19    mode: input::Mode,
20    compressed: input::EntryDataMode,
21    compressed_buf: Option<Vec<u8>>,
22    hash_len: usize,
23    object_hash: gix_hash::Kind,
24}
25
26/// Access
27impl<BR> BytesToEntriesIter<BR> {
28    /// The pack version currently being iterated
29    pub fn version(&self) -> crate::data::Version {
30        self.version
31    }
32
33    /// The kind of iteration
34    pub fn mode(&self) -> input::Mode {
35        self.mode
36    }
37}
38
39/// Initialization
40impl<BR> BytesToEntriesIter<BR>
41where
42    BR: io::BufRead,
43{
44    /// Obtain an iterator from a `read` stream to a pack data file and configure it using `mode` and `compressed`.
45    /// `object_hash` specifies which hash is used for objects in ref-delta entries.
46    ///
47    /// Note that `read` is expected at the beginning of a valid pack data file with a header, entries and a trailer.
48    pub fn new_from_header(
49        mut read: BR,
50        mode: input::Mode,
51        compressed: input::EntryDataMode,
52        object_hash: gix_hash::Kind,
53    ) -> Result<BytesToEntriesIter<BR>, input::Error> {
54        let mut header_data = [0u8; 12];
55        read.read_exact(&mut header_data).map_err(gix_hash::io::Error::from)?;
56
57        let (version, num_objects) = crate::data::header::decode(&header_data)?;
58        assert_eq!(
59            version,
60            crate::data::Version::V2,
61            "let's stop here if we see undocumented pack formats"
62        );
63        Ok(BytesToEntriesIter {
64            read,
65            decompressor: Decompress::new(),
66            compressed,
67            offset: 12,
68            had_error: false,
69            version,
70            objects_left: num_objects,
71            hash: (mode != input::Mode::AsIs).then(|| {
72                let mut hash = gix_hash::hasher(object_hash);
73                hash.update(&header_data);
74                hash
75            }),
76            mode,
77            compressed_buf: None,
78            hash_len: object_hash.len_in_bytes(),
79            object_hash,
80        })
81    }
82
83    fn next_inner(&mut self) -> Result<input::Entry, input::Error> {
84        self.objects_left -= 1; // even an error counts as objects
85
86        // Read header
87        let entry = match self.hash.as_mut() {
88            Some(hash) => {
89                let mut read = read_and_pass_to(
90                    &mut self.read,
91                    HashWrite {
92                        inner: io::sink(),
93                        hash,
94                    },
95                );
96                crate::data::Entry::from_read(&mut read, self.offset, self.hash_len)
97            }
98            None => crate::data::Entry::from_read(&mut self.read, self.offset, self.hash_len),
99        }
100        .map_err(gix_hash::io::Error::from)?;
101
102        // Decompress object to learn its compressed bytes
103        let compressed_buf = self.compressed_buf.take().unwrap_or_else(|| Vec::with_capacity(4096));
104        self.decompressor.reset();
105        let mut decompressed_reader = DecompressRead {
106            inner: read_and_pass_to(
107                &mut self.read,
108                if self.compressed.keep() {
109                    // This buffer stores compressed bytes. The decompressed size is unrelated to the compressed byte
110                    // count and is declared by pack data, so using it as capacity could cause excessive allocation.
111                    // It's notable that `decompressed_size` is also untrusted.
112                    Vec::new()
113                } else {
114                    compressed_buf
115                },
116            ),
117            decompressor: &mut self.decompressor,
118        };
119
120        let bytes_copied = io::copy(&mut decompressed_reader, &mut io::sink()).map_err(gix_hash::io::Error::from)?;
121        if bytes_copied != entry.decompressed_size {
122            return Err(input::Error::IncompletePack {
123                actual: bytes_copied,
124                expected: entry.decompressed_size,
125            });
126        }
127
128        let pack_offset = self.offset;
129        let compressed_size = decompressed_reader.decompressor.total_in();
130        self.offset += entry.header_size() as u64 + compressed_size;
131
132        let mut compressed = decompressed_reader.inner.write;
133        debug_assert_eq!(
134            compressed_size,
135            compressed.len() as u64,
136            "we must track exactly the same amount of bytes as read by the decompressor"
137        );
138        if let Some(hash) = self.hash.as_mut() {
139            hash.update(&compressed);
140        }
141
142        let crc32 = if self.compressed.crc32() {
143            let mut header_buf = [0u8; 12 + gix_hash::Kind::longest().len_in_bytes()];
144            let header_len = entry
145                .header
146                .write_to(bytes_copied, &mut header_buf.as_mut())
147                .map_err(gix_hash::io::Error::from)?;
148            let state = gix_features::hash::crc32_update(0, &header_buf[..header_len]);
149            Some(gix_features::hash::crc32_update(state, &compressed))
150        } else {
151            None
152        };
153
154        let compressed = if self.compressed.keep() {
155            Some(compressed)
156        } else {
157            compressed.clear();
158            self.compressed_buf = Some(compressed);
159            None
160        };
161
162        // Last objects gets trailer (which is potentially verified)
163        let trailer = self.try_read_trailer()?;
164        Ok(input::Entry {
165            header: entry.header,
166            header_size: entry.header_size() as u16,
167            compressed,
168            compressed_size,
169            crc32,
170            pack_offset,
171            decompressed_size: bytes_copied,
172            trailer,
173        })
174    }
175
176    fn try_read_trailer(&mut self) -> Result<Option<ObjectId>, input::Error> {
177        Ok(if self.objects_left == 0 {
178            let mut id = gix_hash::ObjectId::null(self.object_hash);
179            if let Err(err) = self.read.read_exact(id.as_mut_slice()) {
180                if self.mode != input::Mode::Restore {
181                    return Err(input::Error::Io(err.into()));
182                }
183            }
184
185            if let Some(hash) = self.hash.take() {
186                let actual_id = hash.try_finalize().map_err(gix_hash::io::Error::from)?;
187                if self.mode == input::Mode::Restore {
188                    id = actual_id;
189                } else {
190                    actual_id.verify(&id)?;
191                }
192            }
193            Some(id)
194        } else if self.mode == input::Mode::Restore {
195            let hash = self.hash.clone().expect("in restore mode a hash is set");
196            Some(hash.try_finalize().map_err(gix_hash::io::Error::from)?)
197        } else {
198            None
199        })
200    }
201}
202
203fn read_and_pass_to<R: io::Read, W: io::Write>(read: &mut R, to: W) -> PassThrough<&mut R, W> {
204    PassThrough { read, write: to }
205}
206
207impl<R> Iterator for BytesToEntriesIter<R>
208where
209    R: io::BufRead,
210{
211    type Item = Result<input::Entry, input::Error>;
212
213    fn next(&mut self) -> Option<Self::Item> {
214        if self.had_error || self.objects_left == 0 {
215            return None;
216        }
217        let result = self.next_inner();
218        self.had_error = result.is_err();
219        if self.had_error {
220            self.objects_left = 0;
221        }
222        if self.mode == input::Mode::Restore && self.had_error {
223            None
224        } else {
225            Some(result)
226        }
227    }
228
229    fn size_hint(&self) -> (usize, Option<usize>) {
230        (self.objects_left as usize, Some(self.objects_left as usize))
231    }
232}
233
234impl<R> std::iter::ExactSizeIterator for BytesToEntriesIter<R> where R: io::BufRead {}
235
236struct PassThrough<R, W> {
237    read: R,
238    write: W,
239}
240
241impl<R, W> io::BufRead for PassThrough<R, W>
242where
243    Self: io::Read,
244    R: io::BufRead,
245    W: io::Write,
246{
247    fn fill_buf(&mut self) -> io::Result<&[u8]> {
248        self.read.fill_buf()
249    }
250
251    fn consume(&mut self, amt: usize) {
252        let buf = self
253            .read
254            .fill_buf()
255            .expect("never fail as we called fill-buf before and this does nothing");
256        self.write
257            .write_all(&buf[..amt])
258            .expect("a write to never fail - should be a memory buffer");
259        self.read.consume(amt);
260    }
261}
262
263impl<R, W> io::Read for PassThrough<R, W>
264where
265    W: io::Write,
266    R: io::Read,
267{
268    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
269        let bytes_read = self.read.read(buf)?;
270        self.write.write_all(&buf[..bytes_read])?;
271        Ok(bytes_read)
272    }
273}
274
275impl<T> crate::data::File<T>
276where
277    T: crate::FileData,
278{
279    /// Returns an iterator over [`Entries`][crate::data::input::Entry], without making use of the memory mapping.
280    pub fn streaming_iter(&self) -> Result<BytesToEntriesIter<impl io::BufRead>, input::Error> {
281        let reader =
282            io::BufReader::with_capacity(4096 * 8, fs::File::open(&self.path).map_err(gix_hash::io::Error::from)?);
283        BytesToEntriesIter::new_from_header(
284            reader,
285            input::Mode::Verify,
286            input::EntryDataMode::KeepAndCrc32,
287            self.object_hash,
288        )
289    }
290}
291
292/// The boxed variant is faster for what we do (moving the decompressor in and out a lot)
293pub struct DecompressRead<'a, R> {
294    /// The reader from which bytes should be decompressed.
295    pub inner: R,
296    /// The decompressor doing all the work.
297    pub decompressor: &'a mut Decompress,
298}
299
300impl<R> io::Read for DecompressRead<'_, R>
301where
302    R: io::BufRead,
303{
304    fn read(&mut self, into: &mut [u8]) -> io::Result<usize> {
305        gix_zlib::stream::inflate::read(&mut self.inner, self.decompressor, into)
306    }
307}
308
309/// A utility to automatically generate a hash while writing into an inner writer.
310pub struct HashWrite<'a, T> {
311    /// The hash implementation.
312    pub hash: &'a mut Hasher,
313    /// The inner writer.
314    pub inner: T,
315}
316
317impl<T> std::io::Write for HashWrite<'_, T>
318where
319    T: std::io::Write,
320{
321    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
322        let written = self.inner.write(buf)?;
323        self.hash.update(&buf[..written]);
324        Ok(written)
325    }
326
327    fn flush(&mut self) -> std::io::Result<()> {
328        self.inner.flush()
329    }
330}
gix_pack/data/input/bytes_to_entries.rs

gix_pack/data/input/
bytes_to_entries.rs
