Ctx

Ctx<K> passes one field of the broker's per-delivery context to the handler as a parameter.

K is the broker's key for the field, the same key ctx.context(key) reads, so the body needs no &mut Context. Ctx takes any CtxKey: a broker's context key, or MainRuntime. The #[subscriber] attribute infers the context type from the key. A key only reads a context that the broker provides: Ctx<Position> compiles on a retaining memory broker, not on a plain one. Ctx<MainRuntime> is the one key that reads no broker field: it binds the application's runtime for a handler on dedicated threads. Ctx<K> is an extractor, resolved before the body runs.

Example

Macros

examples/context.rs

/// Where an order sits in the broker's log, published on `located`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "located")]
struct Located {
    order: u64,
    sequence: usize,
}

/// `Ctx<Position>` hands the body this delivery's position: one field of the broker's
/// per-delivery context, with no `&mut Context` parameter.
#[subscriber("orders", reply)]
async fn locate(order: &Order, Ctx(at): Ctx<Position>) -> Result<Located, HandlerOutcome> {
    let MemoryPosition::Sequence(sequence) = at else {
        return Err(HandlerOutcome::drop());
    };
    Ok(Located {
        order: order.id,
        sequence,
    })
}

fn app() -> impl App {
    // A retaining broker numbers its messages; its deliveries carry the position.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(16)));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(locate);
    })
}

Manual

examples/manual/context.rs

/// Where an order sits in the broker's log, published on `located`.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Located {
    order: u64,
    sequence: usize,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "located")]` writes.
impl OutgoingDestination for Located {
    type Form = FixedName;

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

impl MessageHeaders for Located {
    type Contract = NoHeaders;
}

/// The broker's per-delivery context is the fourth parameter of `Handle`; the body reads
/// one field of it by its key.
struct Locate;

impl Handle<Order, Located, (), MemoryContext> for Locate {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, MemoryContext>,
    ) -> impl Future<Output = Result<Located, HandlerOutcome>> {
        let MemoryPosition::Sequence(sequence) = ctx.context(Position) else {
            return ready(Err(HandlerOutcome::drop()));
        };
        ready(Ok(Located {
            order: order.id,
            sequence,
        }))
    }
}

fn app() -> impl App {
    // A retaining broker numbers its messages; its deliveries carry the position.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(16)));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Locate).reply().build());
    })
}