pub struct Context<'a, C = (), S = ()> { /* private fields */ }Expand description
Per-delivery context, threaded through middleware and into the handler.
Carries the channel (name), a working copy of the message
headers (middleware may enrich them for the handler; the broker message
itself is untouched), the typed shared application state (where a publisher to
publish from a handler lives), and the broker’s typed per-delivery context read by key
(context / set). The headers copy is made lazily on the first
headers_mut call. Outgoing messages do not inherit it: replies and manual
publishes start from fresh headers, shaped by the publish pipeline.
Implementations§
Source§impl<'a, C, S> Context<'a, C, S>
impl<'a, C, S> Context<'a, C, S>
Sourcepub fn headers_mut(&mut self) -> &mut Headers
pub fn headers_mut(&mut self) -> &mut Headers
The working copy of the message headers, mutably. The first call clones the message headers; later calls return the same copy.
Sourcepub fn state(&self) -> &S
pub fn state(&self) -> &S
Returns the shared application state: the typed S the app’s on_startup produced (or
() when the app declares none), borrowed for the delivery. Read its fields directly.
§Examples
use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerResult};
struct AppState {
prefix: String,
}
async fn handle<M: IncomingMessage>(
_msg: &M,
ctx: &mut Context<'_, (), AppState>,
) -> HandlerResult {
let _prefix = &ctx.state().prefix;
HandlerResult::Ack
}Sourcepub fn context<K: Field<C>>(&self, key: K) -> K::Value<'_>
pub fn context<K: Field<C>>(&self, key: K) -> K::Value<'_>
Reads a broker-supplied per-delivery field off the typed context by compile-time key.
The key is a zero-sized selector the broker exports; resolution is a direct field read off
the typed context (no hashing, boxing, or downcasting). The default () context carries no
fields, so keys exist only for brokers that expose a context type. For shared app state use
state instead.
§Examples
use ruststream::{Field, IncomingMessage};
use ruststream::runtime::{Context, HandlerResult};
// A broker context with one field and the key that reads it.
struct Delivery {
offset: u64,
}
#[derive(Clone, Copy)]
struct Offset;
impl Field<Delivery> for Offset {
type Value<'a> = u64;
fn get(self, d: &Delivery) -> u64 {
d.offset
}
}
async fn handle<M: IncomingMessage>(_m: &M, ctx: &mut Context<'_, Delivery>) -> HandlerResult {
let _offset = ctx.context(Offset);
HandlerResult::Ack
}Sourcepub fn set<K: FieldMut<C>>(&mut self, key: K, value: K::Owned)
pub fn set<K: FieldMut<C>>(&mut self, key: K, value: K::Owned)
Writes a per-delivery scratch value downstream handlers read by key.
Middleware uses this to hand typed data to downstream handlers (an authenticated user, a
correlation id) without serializing it into the headers, when the context type exposes a
writable (FieldMut) key.
§Examples
use ruststream::{Field, FieldMut, IncomingMessage};
use ruststream::runtime::{Context, HandlerResult};
#[derive(Default)]
struct Scratch {
user: Option<u64>,
}
#[derive(Clone, Copy)]
struct User;
impl Field<Scratch> for User {
type Value<'a> = Option<&'a u64>;
fn get(self, s: &Scratch) -> Option<&u64> {
s.user.as_ref()
}
}
impl FieldMut<Scratch> for User {
type Owned = u64;
fn set(self, s: &mut Scratch, value: u64) {
s.user = Some(value);
}
}
async fn handle<M: IncomingMessage>(_m: &M, ctx: &mut Context<'_, Scratch>) -> HandlerResult {
ctx.set(User, 7);
assert_eq!(ctx.context(User), Some(&7));
HandlerResult::Ack
}Sourcepub fn after(&mut self, outcome: HandlerResult) -> After<'_, 'a, C, S>
pub fn after(&mut self, outcome: HandlerResult) -> After<'_, 'a, C, S>
Begins registering a post-settle hook gated on outcome.
The returned builder’s then registers a future that the dispatcher runs
once the message has been settled, but only if the actual settlement matches outcome by
kind. The four kinds are distinct: HandlerResult::Ack, HandlerResult::drop (nack
without requeue), HandlerResult::retry (nack with requeue), and
HandlerResult::retry_after (which matches regardless of the delay). So a hook gated on
drop() does not fire on a retry() settlement, and vice versa. Multiple hooks accumulate
and every matching one runs.
The hook is scoped to the whole delivery. On the batch path a Context is one per batch,
so a hook registered here runs after the entire batch settles; because a batch has
per-element outcomes, the outcome gate is ignored there and only after_settle
hooks (which run regardless) fire (see that method).
§Cancel safety
Post-settle hooks are at-most-once: the message is already settled before any hook runs, so
a hook that panics, or that is lost when the process crashes, never causes a redelivery and
never blocks the next delivery. A graceful shutdown drains in-flight hooks (bounded by the
app’s shutdown_timeout); an aborted shutdown may
drop them.
§Examples
use ruststream::IncomingMessage;
use ruststream::runtime::{Context, Handler, HandlerResult};
fn use_after<M: IncomingMessage + 'static>() {
let _handler = |_msg: &M, ctx: &mut Context| {
ctx.after(HandlerResult::Ack)
.then(async move { /* runs only after this message is acked */ });
async { HandlerResult::Ack }
};
}Sourcepub fn after_ack(&mut self, fut: impl Future<Output = ()> + Send + 'static)
pub fn after_ack(&mut self, fut: impl Future<Output = ()> + Send + 'static)
Registers a post-settle hook that runs only after the message is acked.
Sugar for self.after(HandlerResult::Ack).then(fut); see after for the
gating and cancel-safety semantics.
§Cancel safety
At-most-once, as for after: the ack has already happened, so a lost hook
never redelivers.
§Examples
use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerResult};
fn use_after_ack<M: IncomingMessage + 'static>() {
let _handler = |_msg: &M, ctx: &mut Context| {
ctx.after_ack(async move { /* fire-and-forget once acked */ });
async { HandlerResult::Ack }
};
}Sourcepub fn after_settle(&mut self, fut: impl Future<Output = ()> + Send + 'static)
pub fn after_settle(&mut self, fut: impl Future<Output = ()> + Send + 'static)
Registers a post-settle hook that runs after the message settles, whatever the outcome.
Unlike after this has no outcome gate, so it fires on Ack, Drop,
Retry, and RetryAfter alike. It is the only post-settle form honoured on the batch path, where the
per-element outcomes make an outcome gate ill-defined; there it runs once after the whole
batch has been settled.
§Cancel safety
At-most-once, as for after.
§Examples
use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerResult};
fn use_after_settle<M: IncomingMessage + 'static>() {
let _handler = |_msg: &M, ctx: &mut Context| {
ctx.after_settle(async move { /* runs once the message is settled, any outcome */ });
async { HandlerResult::retry() }
};
}