| @@ -1,25 +1,12 @@ |
| 1 | use crate::data::output; | 1 | use crate::data::input; |
| 2 | use git_features::hash; | 2 | use git_features::hash; |
| 3 | use std::io::Write; | 3 | use std::io::Write; |
| 4 | | 4 | |
| 5 | /// The error returned by `next()` in the [`FromEntriesIter`] iterator. | | |
| 6 | #[allow(missing_docs)] | | |
| 7 | #[derive(Debug, thiserror::Error)] | | |
| 8 | pub enum Error<E> | | |
| 9 | where | | |
| 10 | E: std::error::Error + 'static, | | |
| 11 | { | | |
| 12 | #[error(transparent)] | | |
| 13 | Io(#[from] std::io::Error), | | |
| 14 | #[error(transparent)] | | |
| 15 | Input(E), | | |
| 16 | } | | |
| 17 | | | |
| 18 | /// An implementation of [`Iterator`] to write [encoded entries][output::Entry] to an inner implementation each time | 5 | /// An implementation of [`Iterator`] to write [encoded entries][output::Entry] to an inner implementation each time |
| 19 | /// `next()` is called. | 6 | /// `next()` is called. |
| 20 | pub struct FromEntriesIter<I, W> { | 7 | pub struct FromEntriesIter<I, W> { |
| 21 | /// An iterator for input [`output::Entry`] instances | 8 | /// An iterator for input [`output::Entry`] instances |
| 22 | pub input: I, | 9 | pub inner: I, |
| 23 | /// A way of writing encoded bytes. | 10 | /// A way of writing encoded bytes. |
| 24 | output: hash::Write<W>, | 11 | output: hash::Write<W>, |
| 25 | /// Our trailing hash when done writing all input entries | 12 | /// Our trailing hash when done writing all input entries |
| @@ -27,31 +14,19 @@ pub struct FromEntriesIter<I, W> { · 1 unmodified line |
| 27 | /// The amount of objects in the iteration and the version of the packfile to be written. | 14 | /// The amount of objects in the iteration and the version of the packfile to be written. |
| 28 | /// Will be `None` to signal the header was written already. | 15 | /// Will be `None` to signal the header was written already. |
| 29 | header_info: Option<(crate::data::Version, u32)>, | 16 | header_info: Option<(crate::data::Version, u32)>, |
| 30 | /// The pack data version with which pack entries should be written. | | |
| 31 | entry_version: crate::data::Version, | | |
| 32 | /// The amount of written bytes thus far | | |
| 33 | written: u64, | | |
| 34 | /// Required to quickly find offsets by object IDs, as future objects may refer to those in the past to become a delta offset base. | | |
| 35 | /// It stores the pack offsets at which objects begin. | | |
| 36 | /// Additionally we store if an object was invalid, and if so we will not write it nor will we allow delta objects to it. | | |
| 37 | pack_offsets_and_validity: Vec<(u64, bool)>, | | |
| 38 | /// If we are done, no additional writes will occour | 17 | /// If we are done, no additional writes will occour |
| 39 | is_done: bool, | 18 | is_done: bool, |
| 40 | } | 19 | } |
| 41 | | 20 | |
| 42 | impl<I, W, E> FromEntriesIter<I, W> | 21 | impl<I, W> FromEntriesIter<I, W> |
| 43 | where | 22 | where |
| 44 | I: Iterator<Item = Result<Vec<output::Entry>, E>>, | 23 | I: Iterator<Item = Result<input::Entry, input::Error>>, |
| 45 | W: std::io::Write, | 24 | W: std::io::Write, |
| 46 | E: std::error::Error + 'static, | | |
| 47 | { | 25 | { |
| 48 | /// Create a new instance reading [entries][output::Entry] from an `input` iterator and write pack data bytes to | 26 | /// Create a new instance reading [entries][input::Entry] from an `input` iterator and write pack data bytes to |
| 49 | /// `output` writer, resembling a pack of `version` with exactly `num_entries` amount of objects contained in it. | 27 | /// `output` writer, resembling a pack of `version` with exactly `num_entries` amount of objects contained in it. |
| 50 | /// `hash_kind` is the kind of hash to use for the pack checksum and maybe other places, depending on the version. | 28 | /// `hash_kind` is the kind of hash to use for the pack checksum and maybe other places, depending on the version. |
| 51 | /// | 29 | /// |
| 52 | /// The input chunks are expected to be sorted already. You can use the [InOrderIter][super::InOrderIter] to assure | | |
| 53 | /// this happens on the fly holding entire chunks in memory as long as needed for them to be dispensed in order. | | |
| 54 | /// | | |
| 55 | /// # Panics | 30 | /// # Panics |
| 56 | /// | 31 | /// |
| 57 | /// Not all combinations of `hash_kind` and `version` are supported currently triggering assertion errors. | 32 | /// Not all combinations of `hash_kind` and `version` are supported currently triggering assertion errors. |
| @@ -71,12 +46,9 @@ where · 13 unmodified lines |
| 71 | "currently only Sha1 is supported, right now we don't know how other hashes are encoded", | 46 | "currently only Sha1 is supported, right now we don't know how other hashes are encoded", |
| 72 | ); | 47 | ); |
| 73 | FromEntriesIter { | 48 | FromEntriesIter { |
| 74 | input, | 49 | inner: input, |
| 75 | output: hash::Write::new(output, hash_kind), | 50 | output: hash::Write::new(output, hash_kind), |
| 76 | trailer: None, | 51 | trailer: None, |
| 77 | entry_version: version, | | |
| 78 | pack_offsets_and_validity: Vec::with_capacity(num_entries as usize), | | |
| 79 | written: 0, | | |
| 80 | header_info: Some((version, num_entries)), | 52 | header_info: Some((version, num_entries)), |
| 81 | is_done: false, | 53 | is_done: false, |
| 82 | } | 54 | } |
| @@ -84,75 +56,66 @@ where · 1 unmodified line |
| 84 | | 56 | |
| 85 | /// Consume this instance and return the `output` implementation. | 57 | /// Consume this instance and return the `output` implementation. |
| 86 | /// | 58 | /// |
| 87 | /// _Note_ that the input can be moved out of this instance beforehand. | 59 | /// _Note_ that the `inner` iterator can be moved out of this instance beforehand. |
| 88 | pub fn into_write(self) -> W { | 60 | pub fn into_write(self) -> W { |
| 89 | self.output.inner | 61 | self.output.inner |
| 90 | } | 62 | } |
| 91 | | 63 | |
| 92 | /// Returns the trailing hash over all written entries once done. | 64 | /// Returns the trailing hash over all ~ entries once done. |
| 93 | /// It's `None` if we are not yet done writing. | 65 | /// It's `None` if we are not yet done writing. |
| 94 | pub fn digest(&self) -> Option<git_hash::ObjectId> { | 66 | pub fn digest(&self) -> Option<git_hash::ObjectId> { |
| 95 | self.trailer | 67 | self.trailer |
| 96 | } | 68 | } |
| 97 | | 69 | |
| 98 | fn next_inner(&mut self) -> Result<u64, Error<E>> { | 70 | fn next_inner(&mut self, entry: input::Entry) -> Result<input::Entry, input::Error> { |
| 99 | let previous_written = self.written; | | |
| 100 | if let Some((version, num_entries)) = self.header_info.take() { | 71 | if let Some((version, num_entries)) = self.header_info.take() { |
| 101 | let header_bytes = crate::data::header::encode(version, num_entries); | 72 | let header_bytes = crate::data::header::encode(version, num_entries); |
| 102 | self.output.write_all(&header_bytes[..])?; | 73 | self.output.write_all(&header_bytes[..])?; |
| 103 | self.written += header_bytes.len() as u64; | | |
| 104 | } | 74 | } |
| 105 | match self.input.next() { | 75 | entry |
| 106 | Some(entries) => { | 76 | .header |
| 107 | for entry in entries.map_err(Error::Input)? { | 77 | .write_to(entry.decompressed_size as u64, &mut self.output)? as u64; |
| 108 | if entry.is_invalid() { | 78 | std::io::copy( |
| 109 | self.pack_offsets_and_validity.push((0, false)); | 79 | &mut &*entry |
| 110 | continue; | 80 | .compressed |
| 111 | }; | 81 | .as_ref() |
| 112 | self.pack_offsets_and_validity.push((self.written, true)); | 82 | .map(|v| v.as_slice()) |
| 113 | let header = entry.to_entry_header(self.entry_version, |index| { | 83 | .expect("caller must configure generator to keep compressed bytes"), |
| 114 | let (base_offset, is_valid_object) = self.pack_offsets_and_validity[index]; | 84 | &mut self.output, |
| 115 | if !is_valid_object { | 85 | )?; |
| 116 | unreachable!("if you see this the object database is correct as a delta refers to a non-existing object") | 86 | Ok(entry) |
| 117 | } | 87 | } |
| 118 | self.written - base_offset | 88 | |
| 119 | }); | 89 | fn write_digest(&mut self) -> Result<(), input::Error> { |
| 120 | self.written += header.write_to(entry.decompressed_size as u64, &mut self.output)? as u64; | 90 | let digest = self.output.hash.clone().digest(); |
| 121 | self.written += std::io::copy(&mut &*entry.compressed_data, &mut self.output)? as u64; | 91 | self.output.inner.write_all(&digest[..])?; |
| 122 | } | 92 | self.output.inner.flush()?; |
| 123 | } | 93 | self.is_done = true; |
| 124 | None => { | 94 | self.trailer = Some(git_hash::ObjectId::from(digest)); |
| 125 | let digest = self.output.hash.clone().digest(); | 95 | Ok(()) |
| 126 | self.output.inner.write_all(&digest[..])?; | | |
| 127 | self.written += digest.len() as u64; | | |
| 128 | self.output.inner.flush()?; | | |
| 129 | self.is_done = true; | | |
| 130 | self.trailer = Some(git_hash::ObjectId::from(digest)); | | |
| 131 | } | | |
| 132 | }; | | |
| 133 | Ok(self.written - previous_written) | | |
| 134 | } | 96 | } |
| 135 | } | 97 | } |
| 136 | | 98 | |
| 137 | impl<I, W, E> Iterator for FromEntriesIter<I, W> | 99 | impl<I, W> Iterator for FromEntriesIter<I, W> |
| 138 | where | 100 | where |
| 139 | I: Iterator<Item = Result<Vec<output::Entry>, E>>, | 101 | I: Iterator<Item = Result<input::Entry, input::Error>>, |
| 140 | W: std::io::Write, | 102 | W: std::io::Write, |
| 141 | E: std::error::Error + 'static, | | |
| 142 | { | 103 | { |
| 143 | /// The amount of bytes written to `out` if `Ok` or the error `E` received from the input. | 104 | /// The amount of bytes written to `out` if `Ok` or the error `E` received from the input. |
| 144 | type Item = Result<u64, Error<E>>; | 105 | type Item = Result<input::Entry, input::Error>; |
| 145 | | 106 | |
| 146 | fn next(&mut self) -> Option<Self::Item> { | 107 | fn next(&mut self) -> Option<Self::Item> { |
| 147 | if self.is_done { | 108 | if self.is_done { |
| 148 | return None; | 109 | return None; |
| 149 | } | 110 | } |
| 150 | Some(match self.next_inner() { | 111 | |
| 151 | Err(err) => { | 112 | match self.inner.next() { |
| | 113 | Some(Ok(entry)) => Some(self.next_inner(entry)), |
| | 114 | Some(Err(err)) => { |
| 152 | self.is_done = true; | 115 | self.is_done = true; |
| 153 | Err(err) | 116 | Some(Err(err)) |
| 154 | } | 117 | } |
| 155 | Ok(written) => Ok(written), | 118 | None => self.write_digest().err().map(Err), |
| 156 | }) | 119 | } |
| 157 | } | 120 | } |
| 158 | } | 121 | } |