| 70 | type Item = Result<Event, anyhow::Error>; |
| 71 | |
| 72 | fn next(&mut self) -> Option<Self::Item> { |
| 73 | self.position = self.source.stream_position().unwrap(); |
| 74 | if self.position >= self.size { |
| 75 | return None; |
| 76 | } |
| 77 | match EventSerializationConfig::config().deserialize_from(&mut self.source) { |
| 78 | Ok(event) => Some(Ok(event)), |
| 79 | Err(error) => match error.deref() { |
| 80 | bincode::ErrorKind::Io(e) |
| 81 | if matches!(e.kind(), std::io::ErrorKind::UnexpectedEof) => |
| 82 | { |
| 83 | self.partial_data_error = true; |
| 84 | None |
| 85 | } |
| 86 | _ => Some(Err(anyhow!( |
| 87 | "Load journal event failed at position: {}. Deserialization error: {error:?}", |
| 88 | self.position |
| 89 | ))), |
| 90 | }, |
| 91 | } |
| 92 | } |
| 93 | } |
| 94 | |
| 95 | #[cfg(test)] |