FileBatchContext

A batch body reaches the seek handle through FileBatchContext, which holds no position; each element carries its own sequence.

A batch spans many deliveries, so it has no single place in the file. The batch context holds what every delivery of the batch shares: the subscription's SeekHandle. A batch body takes ctx: &mut Context<'_, FileBatchContext> and reads the handle with ctx.context(SeekHandle). The batch settles first, so a seek from its body applies to the batches after it. The deliveries the current batch already holds stay in it. Each element's own sequence is in its stream-sequence header (SEQUENCE_HEADER). Position is a key of the per-delivery FileContext, which a single-message handler reads. In the manual form the body is an impl Handle<[T], (), (), FileBatchContext>.

Example

Job 1 says the jobs up to sequence 5 cannot be processed: the first batch holds jobs 1 to 3, and the next one opens at job 5.

Macros
Manual

examples/file_batch_seek.rs
/// A batch spans many deliveries, so its context is the subscription's: the seek handle, and no
/// position. The move applies to the batches after this one.
#[subscriber(FileStream::new("jobs"), start_at(FilePosition::beginning()))]
async fn drain(jobs: &[Job], ctx: &mut Context<'_, FileBatchContext>) -> HandlerOutcome {
    tracing::info!(jobs = jobs.len(), "drained");
    let Some(resume_at) = jobs.iter().find_map(|job| job.resume_at) else {
        return HandlerOutcome::ack();
    };
    if let Err(error) = ctx
        .context(SeekHandle)
        .seek(FilePosition::sequence(resume_at))
        .await
    {
        tracing::warn!(resume_at, %error, "seek refused");
    }
    HandlerOutcome::ack()
}

/// The worker on the job file `main` hands it, three jobs at a time.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(file, |b| {
        b.include(drain.batch(nonzero!(3)));
    })
}

examples/manual/file_batch_seek.rs
/// A batch spans many deliveries, so its context is the subscription's: the seek handle, and no
/// position. The move applies to the batches after this one.
struct Drain;

impl Handle<[Job], (), (), FileBatchContext> for Drain {
    async fn handle(
        &self,
        jobs: &[Job],
        _outs: &(),
        ctx: &mut Context<'_, FileBatchContext>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        tracing::info!(jobs = jobs.len(), "drained");
        let Some(resume_at) = jobs.iter().find_map(|job| job.resume_at) else {
            return Ok(());
        };
        if let Err(error) = ctx
            .context(SeekHandle)
            .seek(FilePosition::sequence(resume_at))
            .await
        {
            tracing::warn!(resume_at, %error, "seek refused");
        }
        Ok(())
    }
}

/// The worker on the job file `main` hands it, three jobs at a time.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(file, |b| {
        b.include(
            subscriber(FileStream::new("jobs"), Drain)
                .batch(nonzero!(3))
                .start_at(FilePosition::beginning())
                .build(),
        );
    })
}

Example​

Example