Parse [`Header`] from `buf`, returning the number of bytes read This method can be called multiple times with consecutive chunks of data, allowing integration with chunked IO systems like [`BufRead::fill_buf`] All errors should be considered fatal, and decoding aborted Once the entire [`Header`] has been decoded this method will not read any further input bytes, and the header can be obtained w
(&mut self, mut buf: &[u8])
| 237 | /// |
| 238 | /// [`BufRead::fill_buf`]: std::io::BufRead::fill_buf |
| 239 | pub fn decode(&mut self, mut buf: &[u8]) -> Result<usize, AvroError> { |
| 240 | let max_read = buf.len(); |
| 241 | while !buf.is_empty() { |
| 242 | match self.state { |
| 243 | HeaderDecoderState::Magic => { |
| 244 | let remaining = &MAGIC[MAGIC.len() - self.bytes_remaining..]; |
| 245 | let to_decode = buf.len().min(remaining.len()); |
| 246 | if !buf.starts_with(&remaining[..to_decode]) { |
| 247 | return Err(AvroError::ParseError("Incorrect avro magic".to_string())); |
| 248 | } |
| 249 | self.bytes_remaining -= to_decode; |
| 250 | buf = &buf[to_decode..]; |
| 251 | if self.bytes_remaining == 0 { |
| 252 | self.state = HeaderDecoderState::BlockCount; |
| 253 | } |
| 254 | } |
| 255 | HeaderDecoderState::BlockCount => { |
| 256 | if let Some(block_count) = self.vlq_decoder.long(&mut buf) { |
| 257 | match block_count.try_into() { |
| 258 | Ok(0) => { |
| 259 | self.state = HeaderDecoderState::Sync; |
| 260 | self.bytes_remaining = 16; |
| 261 | } |
| 262 | Ok(remaining) => { |
| 263 | self.tuples_remaining = remaining; |
| 264 | self.state = HeaderDecoderState::KeyLen; |
| 265 | } |
| 266 | Err(_) => { |
| 267 | self.tuples_remaining = block_count.unsigned_abs() as _; |
| 268 | self.state = HeaderDecoderState::BlockLen; |
| 269 | } |
| 270 | } |
| 271 | } |
| 272 | } |
| 273 | HeaderDecoderState::BlockLen => { |
| 274 | if self.vlq_decoder.long(&mut buf).is_some() { |
| 275 | self.state = HeaderDecoderState::KeyLen |
| 276 | } |
| 277 | } |
| 278 | HeaderDecoderState::Key => { |
| 279 | let to_read = self.bytes_remaining.min(buf.len()); |
| 280 | self.meta_buf.extend_from_slice(&buf[..to_read]); |
| 281 | self.bytes_remaining -= to_read; |
| 282 | buf = &buf[to_read..]; |
| 283 | if self.bytes_remaining == 0 { |
| 284 | self.meta_offsets.push(self.meta_buf.len()); |
| 285 | self.state = HeaderDecoderState::ValueLen; |
| 286 | } |
| 287 | } |
| 288 | HeaderDecoderState::Value => { |
| 289 | let to_read = self.bytes_remaining.min(buf.len()); |
| 290 | self.meta_buf.extend_from_slice(&buf[..to_read]); |
| 291 | self.bytes_remaining -= to_read; |
| 292 | buf = &buf[to_read..]; |
| 293 | if self.bytes_remaining == 0 { |
| 294 | self.meta_offsets.push(self.meta_buf.len()); |
| 295 | |
| 296 | self.tuples_remaining -= 1; |