Function commits

crates/commitlog/src/stream/reader.rs:41–62

Stream the `range` of transaction offsets from the commitlog in `repo` as raw commitlog data. The stream contains segment headers as they are encountered scanning the `range`. Only whole [`commit::StoredCommit`]s are yielded, so a `range` that doesn't fall on commit boundaries may yield extra data. Only the headers of the source commitlog are inspected (in order to be able to satisfy the `range

(repo: R, range: impl RangeBounds<u64>)

/// If the commitlog is empty, that is does not contain any commits, the
/// returned stream yields nothing.
pub fn commits<R>(repo: R, range: impl RangeBounds<u64>) -> impl Stream<Item = io::Result<Bytes>>
where
    R: AsyncRepo + Send + 'static,
{
    let mut range = RangeFromMaybeToInclusive::from_range_bounds(range);
    let retain = move |segments: Vec<_>| retain_range(&segments, range);
    try_stream! {
        let segments = repo.existing_offsets().map(retain)?;
        for segment_offset in segments {
            if range.start < segment_offset {
                range.start = segment_offset;
            }
            trace!("segment: segment={} start={}", segment_offset, range.start);

            let segment = repo.open_segment_reader_async(segment_offset).await?;

            for await chunk in read_segment(repo.clone(), segment, segment_offset, range) {
                yield chunk.inspect_err(|e| warn!("error reading segment {segment_offset}: {e}"))?;
            }
        }
    }
}

fn read_segment(
    repo: impl Repo + Send + 'static,