Context

Context carries one delivery's channel, headers, state and broker fields to the middleware and the handler.

The runtime builds one Context per delivery and lends it to the middleware and then to the handler. A #[subscriber] function takes it as &mut Context<'_> after the message; a manual handler gets it as the last argument of handle. name() is the channel the message arrived on. headers() reads the delivery's headers; headers_mut() gives a working copy that middleware can enrich for the handler. The working copy stays inside the delivery: a reply or a publish from the handler starts with fresh headers. state() is the application state that on_startup produced. The context's first type parameter is the broker's per-delivery context: () for most subscriptions, MemoryContext for a retaining memory broker. The broker exports keys for its fields, and ctx.context(key) reads one; Ctx<K> does the same read as a handler parameter. A key implements ContextField, which names the context it reads, and Field, which reads the value out of it with no lookup at run time. The memory broker's seek handle is one such field. The context also schedules work for after the settlement: the post-settle hooks.

Example

Macros
Manual

examples/context.rs
/// What the service read off one delivery, published on `receipts`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
    channel: String,
    request: Option<String>,
}

#[subscriber("orders", reply)]
async fn receive(order: &Order, ctx: &mut Context<'_>) -> Receipt {
    let request = ctx
        .headers()
        .get("x-request-id")
        .map(|id| String::from_utf8_lossy(id).into_owned());
    Receipt {
        order: order.id,
        channel: ctx.name().to_owned(),
        request,
    }
}

examples/manual/context.rs
/// What the service read off one delivery, published on `receipts`.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Receipt {
    order: u64,
    channel: String,
    request: Option<String>,
}

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

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

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

/// The context is the last parameter of `handle`, always in reach.
struct Receive;

impl Handle<Order, Receipt> for Receive {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        let request = ctx
            .headers()
            .get("x-request-id")
            .map(|id| String::from_utf8_lossy(id).into_owned());
        ready(Ok(Receipt {
            order: order.id,
            channel: ctx.name().to_owned(),
            request,
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Receive).reply().build());
    })
}

Example​

Example