seek

A handler moves its own subscription to another position of the log through the broker's seek handle.

The in-memory broker's per-delivery context carries the handle under the key SeekHandle. A handler reads it with the Ctx<SeekHandle> parameter; the manual form declares MemoryContext on its Handle and reads ctx.context(SeekHandle). seek takes a MemoryPosition, the same type start_at takes. Once seek resolves, the next delivery comes from the new position, and entries already queued before it are skipped. A sequence the retention bound has already evicted is refused with an error, and the example retries the delivery then. The handle implements Seeker, the trait a broker with a replayable log provides. Ctx<SeekHandle> does not compile against MemoryBroker::new(), which keeps no log.

Example

The handler meets a poison marker and moves its subscription past the entries behind it.

Macros
Manual

examples/seek.rs
/// The id that marks a poisoned region: the entries up to the resume point are skipped.
const POISON: u64 = 999;

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "jobs")]
struct Job {
    id: u64,
}

/// The seek handle is a field of the broker's per-delivery context, read with `Ctx`.
#[subscriber("jobs", start_at(MemoryPosition::start()))]
async fn work(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
    if job.id == POISON {
        // Resume at the fourth entry of the log.
        if seeker.seek(MemoryPosition::sequence(3)).await.is_err() {
            return HandlerOutcome::retry();
        }
        return HandlerOutcome::ack();
    }
    tracing::info!(job.id, "processed");
    HandlerOutcome::ack()
}

examples/manual/seek.rs
/// The id that marks a poisoned region: the entries up to the resume point are skipped.
const POISON: u64 = 999;

#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Job {
    id: u64,
}

// `#[outgoing(name = "jobs")]` by hand: the type fixes its destination.
impl OutgoingDestination for Job {
    type Form = FixedName;

    const DESTINATION: &'static str = "jobs";
}

impl MessageHeaders for Job {
    type Contract = NoHeaders;
}

struct Work;

// Declaring the broker's `MemoryContext` is what puts the seek handle in the context.
impl Handle<Job, (), (), MemoryContext> for Work {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, MemoryContext>,
    ) -> Result<(), HandlerOutcome> {
        if job.id == POISON {
            // Resume at the fourth entry of the log.
            let seeker = ctx.context(SeekHandle);
            if seeker.seek(MemoryPosition::sequence(3)).await.is_err() {
                return Err(HandlerOutcome::retry());
            }
            return Ok(());
        }
        tracing::info!(job.id, "processed");
        Ok(())
    }
}

fn app(broker: MemoryBroker<Retaining>) -> impl App {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber("jobs", Work)
                .start_at(MemoryPosition::start())
                .build(),
        );
    })
}

Example​

Example