PulsarBatchContext

A batch handler's context, PulsarBatchContext, carries the subscription's seek handle alone; asking a batch for Position does not compile.

A batch spans many deliveries, so it has no single position in the log. The subscription builds the context once per batch, from what every delivery of the batch shares: the SeekHandle key, the same key a PulsarContext carries. A batch body takes ctx: &mut Context<'_, PulsarBatchContext> and reads the handle with ctx.context(SeekHandle). A seek from a batch body takes effect once the batch has settled, so the next batch opens at the new position. Position is a compile error in a batch body, because the batch context does not carry it. In the manual form the context is the fourth parameter of Handle: impl Handle<[Job], (), (), PulsarBatchContext>.

Example

A batch that holds a job whose result was lost replays the retained log once.

Macros

examples/pulsar_context.rs

/// A batch with a job whose result was lost settles first; the seek then replays the
/// retained log, so the next batch opens at the beginning.
#[subscriber(PulsarSubscription::new("jobs", "bulk"))]
async fn work(
    jobs: &[Job],
    ctx: &mut Context<'_, PulsarBatchContext>,
    State(redos): State<Arc<Redos>>,
) -> HandlerOutcome {
    tracing::info!(count = jobs.len(), "jobs done");
    let redo = jobs.iter().any(|job| job.redo && redos.first_time(job));
    if redo
        && ctx
            .context(SeekHandle)
            .seek(PulsarPosition::earliest())
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/pulsar_context.rs

/// A batch with a job whose result was lost settles first; the seek then replays the
/// retained log, so the next batch opens at the beginning.
struct Work;

impl Handle<[Job], (), (), PulsarBatchContext, AppState> for Work {
    async fn handle(
        &self,
        jobs: &[Job],
        _outs: &(),
        ctx: &mut Context<'_, PulsarBatchContext, AppState>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        tracing::info!(count = jobs.len(), "jobs done");
        let redos = &ctx.state().redos;
        let redo = jobs.iter().any(|job| job.redo && redos.first_time(job));
        if redo
            && ctx
                .context(SeekHandle)
                .seek(PulsarPosition::earliest())
                .await
                .is_err()
        {
            return Err(jobs.iter().map(|_| HandlerOutcome::retry()).collect());
        }
        Ok(())
    }
}