Struct Context 

pub struct Context<'a, C = (), S = ()> { /* private fields */ }

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 delivery’s map is taken off the broker on the first read of it, and the working copy is made 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.

Examples

A handler reads its delivery through the context and takes extractors after it:

use ruststream::memory::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[derive(Clone)]
struct Region(String);

#[derive(Clone, FromRef)]
struct AppState {
    region: Region,
}

#[subscriber("orders", workers(4))]
async fn handle(
    order: &Order,
    ctx: &mut Context<'_, (), AppState>,
    State(region): State<Region>,
) -> HandlerOutcome {
    if ctx.headers().get("x-dry-run").is_some() {
        return HandlerOutcome::ack();
    }
    println!("order {} for {} on {}", order.id, region.0, ctx.name());
    HandlerOutcome::ack()
}

Implementations

impl<'a, C, S> Context<'a, C, S>

pub fn decode_policy(&self) -> FailurePolicy

The materialization policy in force for this delivery: what happens to a payload that does not decode, or a header contract that does not parse.

The #[subscriber] expansion reads it to settle a failed Headers extraction; a hand-written handler can read it for the same reason.

Examples

use std::str;

use ruststream::IncomingMessage;
use ruststream::runtime::{Context, FailurePolicy, HandlerOutcome};

/// A hand-written handler reads the payload as text itself, and settles one that is not the
/// way the registration's decode policy says.
async fn handle<M: IncomingMessage>(msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
    let Ok(line) = str::from_utf8(msg.payload()) else {
        return match ctx.decode_policy() {
            FailurePolicy::Retry => HandlerOutcome::retry(),
            FailurePolicy::RetryAfter(delay) => HandlerOutcome::retry_after(delay),
            FailurePolicy::Skip => HandlerOutcome::ack(),
            _ => HandlerOutcome::drop(),
        };
    };
    tracing::info!(line, "handled");
    HandlerOutcome::ack()
}

pub fn name(&self) -> &str

The channel (name / subject) the message arrived on.

pub fn main_runtime(&self) -> &MainRuntime

The runtime the app runs on: the one explicit way from a handler on dedicated threads (threads(n)) to send work to the app’s runtime (see MainRuntime). On any other placement the handler already runs there.

Examples

use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerOutcome};

async fn notify(_id: u64) {}

async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
    // The notification leaves the handler's thread; the handler does not wait for it.
    ctx.main_runtime().spawn(notify(7));
    HandlerOutcome::ack()
}

pub fn headers(&self) -> &HeaderMap

The working copy of the message headers.

pub fn headers_mut(&mut self) -> &mut HeaderMap

The working copy of the message headers, mutably. The first call clones the message headers; later calls return the same copy.

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::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

/// What `on_startup` produced, shared by every delivery.
struct Config {
    region: String,
}

#[subscriber("orders")]
async fn handle(order: &Order, ctx: &mut Context<'_, (), Config>) -> HandlerOutcome {
    tracing::info!(order.id, region = ctx.state().region, "accepted");
    HandlerOutcome::ack()
}

pub fn context<K>(&self, key: K) -> <K as Field<C>>::Value<'_>

where K: Field<C>,

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::runtime::{Context, HandlerOutcome};
use ruststream::{Field, IncomingMessage};

/// A broker's per-delivery 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>(
    _msg: &M,
    ctx: &mut Context<'_, Delivery>,
) -> HandlerOutcome {
    tracing::info!(offset = ctx.context(Offset), "handled");
    HandlerOutcome::ack()
}

pub fn set<K>(&mut self, key: K, value: <K as FieldMut<C>>::Owned)

where K: FieldMut<C>,

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::runtime::{Context, HandlerOutcome};
use ruststream::{Field, FieldMut, IncomingMessage};

/// A broker's per-delivery context, with a slot for the authenticated user.
#[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);
    }
}

/// Verifies the delivery's bearer token and records whose it is, for every later stage
/// that reads `User`.
async fn authenticate<M: IncomingMessage>(
    _msg: &M,
    ctx: &mut Context<'_, Scratch>,
) -> HandlerOutcome {
    let Some(user) = ctx
        .headers()
        .get_str("authorization")
        .and_then(|value| value.strip_prefix("Bearer "))
        .and_then(verify_token)
    else {
        return HandlerOutcome::drop();
    };
    ctx.set(User, user);
    HandlerOutcome::ack()
}

pub fn after(&mut self, outcome: HandlerOutcome) -> 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: HandlerOutcome::ack, HandlerOutcome::drop (nack without requeue), HandlerOutcome::retry (nack with requeue), and HandlerOutcome::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. Any continuation already attached to the gate value itself is ignored: the hook future is what then registers.

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::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

async fn notify_shipping(_id: u64) {}

#[subscriber("orders")]
async fn accept(order: &Order, ctx: &mut Context<'_>) -> HandlerOutcome {
    let id = order.id;
    // Shipping hears of the order only once it is acked, never for one that comes back.
    ctx.after(HandlerOutcome::ack())
        .then(async move { notify_shipping(id).await });
    HandlerOutcome::ack()
}

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(HandlerOutcome::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::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

async fn warm_cache(_id: u64) {}

#[subscriber("orders")]
async fn accept(order: &Order, ctx: &mut Context<'_>) -> HandlerOutcome {
    let id = order.id;
    ctx.after_ack(async move { warm_cache(id).await });
    HandlerOutcome::ack()
}

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 std::time::Instant;

use ruststream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn accept(order: &Order, ctx: &mut Context<'_>) -> HandlerOutcome {
    let started = Instant::now();
    // Recorded whatever the outcome: an ack, a drop or a retry.
    ctx.after_settle(async move {
        tracing::info!(elapsed = ?started.elapsed(), "settled");
    });
    if order.id == 0 {
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

Trait Implementations

impl<C, S> Debug for Context<'_, C, S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.