(
&mut self,
transactions: impl IntoIterator<Item = T>,
)
| 114 | |
| 115 | impl<W: io::Write> Writer<W> { |
| 116 | pub fn commit<T: Into<Transaction<U>>, U: Encode>( |
| 117 | &mut self, |
| 118 | transactions: impl IntoIterator<Item = T>, |
| 119 | ) -> io::Result<Option<Committed>> { |
| 120 | for tx in transactions { |
| 121 | let tx = tx.into(); |
| 122 | let expected_offset = self.commit.min_tx_offset + self.commit.n as u64; |
| 123 | if tx.offset != expected_offset { |
| 124 | self.commit.n = 0; |
| 125 | self.commit.records.clear(); |
| 126 | |
| 127 | return Err(io::Error::new( |
| 128 | io::ErrorKind::InvalidInput, |
| 129 | format!("invalid transaction offset {}, expected {}", tx.offset, expected_offset), |
| 130 | )); |
| 131 | } |
| 132 | assert!( |
| 133 | self.commit.n < u16::MAX, |
| 134 | "maximum number of transactions in a single commit exceeded" |
| 135 | ); |
| 136 | self.commit.n += 1; |
| 137 | tx.txdata.encode_record(&mut self.commit.records); |
| 138 | } |
| 139 | |
| 140 | if self.commit.n == 0 { |
| 141 | return Ok(None); |
| 142 | } |
| 143 | |
| 144 | let checksum = self |
| 145 | .commit |
| 146 | .write(&mut self.inner) |
| 147 | // Panic here as we don't know how much of the commit has been |
| 148 | // written (if anything). Further commits would leave corrupted data |
| 149 | // in the log. |
| 150 | .unwrap_or_else(|e| panic!("failed to write commit {}: {:#}", self.commit.min_tx_offset, e)); |
| 151 | let commit_len = self.commit.encoded_len() as u64; |
| 152 | |
| 153 | if let Some(index) = self.offset_index_head.as_mut() { |
| 154 | let _ = index |
| 155 | .append_after_commit(self.commit.min_tx_offset, self.bytes_written, commit_len) |
| 156 | .inspect_err(|e| debug!("failed to append to offset index: {e}")); |
| 157 | } |
| 158 | |
| 159 | let tx_range_start = self.commit.min_tx_offset; |
| 160 | |
| 161 | self.bytes_written += commit_len; |
| 162 | self.commit.min_tx_offset += self.commit.n as u64; |
| 163 | self.commit.n = 0; |
| 164 | self.commit.records.clear(); |
| 165 | |
| 166 | Ok(Some(Committed { |
| 167 | tx_range: tx_range_start..self.commit.min_tx_offset, |
| 168 | checksum, |
| 169 | })) |
| 170 | } |
| 171 | |
| 172 | pub fn flush(&mut self) -> io::Result<()> { |
| 173 | self.inner.flush() |
nothing calls this directly
no test coverage detected