Seek

Ctx<SeekHandle> moves the running subscription; what was queued is discarded and a position past the tip is refused.

Ctx(seeker): Ctx<SeekHandle> binds a FileSeeker for the subscription the delivery came from (SeekHandle). Its seek(position) takes a FilePosition, and the next delivery comes from there. Deliveries already queued from the old place are discarded. Seek to a position the file has reached. The sequence the file writes next is the tip, and a seek there succeeds. A sequence further ahead, and an instant later than every message the file holds, are refused with an error. A live subscription then ends, and a replay stays where it was. A position read with Ctx<Position>, the beginning and the tip are always within the file. A seek after the broker has shut down returns an error. A batch body reaches the same handle through FileBatchContext. In the manual form the body names FileContext as its context and reads ctx.context(SeekHandle). A position that has to hold across a restart belongs in start_at.

Example

Job 2 says the jobs up to sequence 4 cannot be processed, so the subscription jumps past job 3.

Macros
Manual

examples/file_seek.rs
/// Skips a run of jobs known to fail: the subscription jumps to `resume_at`.
#[subscriber(FileStream::new("jobs"), start_at(FilePosition::beginning()))]
async fn work(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
    tracing::info!(job.id, "worked");
    let Some(resume_at) = job.resume_at else {
        return HandlerOutcome::ack();
    };
    if let Err(error) = seeker.seek(FilePosition::sequence(resume_at)).await {
        tracing::warn!(job.id, resume_at, %error, "seek refused");
    }
    HandlerOutcome::ack()
}

/// The worker on the job file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(file, |b| {
        b.include(work);
    })
}

examples/manual/file_seek.rs
/// Skips a run of jobs known to fail: the subscription jumps to `resume_at`.
struct Work;

impl Handle<Job, (), (), FileContext> for Work {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, FileContext>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(job.id, "worked");
        let Some(resume_at) = job.resume_at else {
            return Ok(());
        };
        let seeker = ctx.context(SeekHandle);
        if let Err(error) = seeker.seek(FilePosition::sequence(resume_at)).await {
            tracing::warn!(job.id, resume_at, %error, "seek refused");
        }
        Ok(())
    }
}

/// The worker on the job file `main` hands it.
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"), Work)
                .start_at(FilePosition::beginning())
                .build(),
        );
    })
}

Example​

Example