Seek

A handler moves its whole subscription through the SeekHandle key; queued deliveries are dropped and replay starts at the target.

Ctx<SeekHandle> binds a PulsarSeeker for the subscription the delivery came from. Its seek(position) takes a PulsarPosition, and the next delivery comes from there. The seek moves the subscription, not one consumer: every topic and every partition it reads, and every consumer that shares it. Deliveries already queued when the seek lands are dropped, so the handler never sees them. A batch still being assembled is the exception: it closes with the deliveries it had taken. A seeker is a value, so a handler can keep it and move the subscription later from elsewhere, as the PulsarPosition example does. A seek after the service shut down returns PulsarError::NotConnected. A batch handler reaches the same key through its PulsarBatchContext. A position that has to apply on each boot belongs in start_at instead.

Replay and seeking indexes every order again with a seek from the handler.

Example

A job marked poisoned starts a run the producer knows to be unprocessable; the subscription jumps to the tip of the log past it.

Macros
Manual

examples/pulsar_positions_seek.rs
/// Skips a poisoned run: the subscription jumps to the tip of the log, and the jobs queued
/// behind the marker never reach the handler.
#[subscriber(PulsarSubscription::new("jobs", "workers"))]
async fn work(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
    tracing::info!(job.id, "worked");
    if !job.poisoned {
        return HandlerOutcome::ack();
    }
    match seeker.seek(PulsarPosition::latest()).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(job.id, %err, "the poisoned run cannot be skipped");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/pulsar_positions_seek.rs
/// Skips a poisoned run: the subscription jumps to the tip of the log, and the jobs queued
/// behind the marker never reach the handler.
struct Work;

impl Handle<Job, (), (), PulsarContext> for Work {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, PulsarContext>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(job.id, "worked");
        if !job.poisoned {
            return Ok(());
        }
        ctx.context(SeekHandle)
            .seek(PulsarPosition::latest())
            .await
            .map_err(|err| {
                tracing::warn!(job.id, %err, "the poisoned run cannot be skipped");
                HandlerOutcome::retry()
            })
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(subscriber(PulsarSubscription::new("jobs", "workers"), Work).build());
        },
    )
}

Example​

Example