PulsarContext

PulsarContext gives a handler where its delivery sits in the log and the handle that moves the subscription, each read through Ctx.

The subscription builds it once per delivery, beside the payload and the headers, with no allocation. A handler binds one field with Ctx<K>, where K is a key:

Key	Value	What it is
`Position`	`PulsarPosition`	the delivery's message id; a seek to it delivers this message again
`SeekHandle`	`PulsarSeeker`	the handle that moves the subscription, on every topic it reads

Both keys are in the prelude. Ctx<Position> copies the message id, while ctx.context(Position) borrows it. A body that reads both keys takes ctx: &mut Context<'_, PulsarContext> instead and calls ctx.context(key) per key. In the manual form the context is the fourth parameter of Handle: impl Handle<Job, (), (), PulsarContext>. A batch body reads a PulsarBatchContext, which carries the seek handle alone.

Replay and seeking reads the seek handle from the context to index every order again.

Example

A job whose result was lost moves the subscription back to its own position, so it is done again.

Macros
Manual

examples/pulsar_context.rs
/// Logs where each job sits; a job whose result was lost moves the subscription back to its
/// own position, so the job is delivered again.
#[subscriber(PulsarSubscription::new("jobs", "workers"))]
async fn work(
    job: &Job,
    ctx: &mut Context<'_, PulsarContext>,
    State(redos): State<Arc<Redos>>,
) -> HandlerOutcome {
    tracing::info!(job.id, at = ?ctx.context(Position), "job done");
    if job.redo && redos.first_time(job) {
        let at = ctx.context(Position).clone();
        if ctx.context(SeekHandle).seek(at).await.is_err() {
            return HandlerOutcome::retry();
        }
    }
    HandlerOutcome::ack()
}

examples/manual/pulsar_context.rs
/// Logs where each job sits; a job whose result was lost moves the subscription back to its
/// own position, so the job is delivered again.
struct Work;

impl Handle<Job, (), (), PulsarContext, AppState> for Work {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, PulsarContext, AppState>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(job.id, at = ?ctx.context(Position), "job done");
        if job.redo && ctx.state().redos.first_time(job) {
            let at = ctx.context(Position).clone();
            if ctx.context(SeekHandle).seek(at).await.is_err() {
                return Err(HandlerOutcome::retry());
            }
        }
        Ok(())
    }
}

Example​

Example