Extractors

A type that implements FromContext becomes a handler parameter, resolved from the delivery before the body runs.

Every parameter of a #[subscriber] function after the message and the optional &mut Context is an extractor. The generated handler resolves the extractors in the order they are declared, then runs the body. from_context reads the context mutably, so an extractor can read headers, broker fields and the state. It is async, so it may also do I/O, such as a lookup. An error is the Rejection, which converts into a HandlerOutcome: the delivery settles with that outcome, and the body does not run. HandlerOutcome itself is a valid rejection; an extractor that cannot fail uses Infallible. A manual handler calls from_context at the top of handle and returns a rejection as its error. The impl is generic over the context type C and the state S, so the extractor works on any subscription. An extractor that reads a broker field or a typed state implements the trait for those types instead. State<T>, the extractor of the application state, and Ctx<K> are extractors of this kind that the crate provides.

Example

Macros

examples/from_context.rs

/// The tenant an order belongs to, read off its `x-tenant` header.
struct Tenant(String);

impl<C: Send, S: Sync> FromContext<C, S> for Tenant {
    type Rejection = HandlerOutcome;

    fn from_context(
        ctx: &mut Context<'_, C, S>,
    ) -> impl Future<Output = Result<Self, HandlerOutcome>> + Send {
        let tenant = ctx.headers().get("x-tenant");
        let tenant = tenant.map(|tenant| Self(String::from_utf8_lossy(tenant).into_owned()));
        // No tenant, no work: the delivery is dropped and the body does not run.
        ready(tenant.ok_or_else(HandlerOutcome::drop))
    }
}

/// An accepted order and its tenant, published on `accepted`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "accepted")]
struct Accepted {
    order: u64,
    tenant: String,
}

/// The tenant arrives as a parameter, resolved before the body runs.
#[subscriber("orders", reply)]
async fn accept(order: &Order, Tenant(tenant): Tenant) -> Accepted {
    Accepted {
        order: order.id,
        tenant,
    }
}

Manual

examples/manual/from_context.rs

/// The tenant an order belongs to, read off its `x-tenant` header.
struct Tenant(String);

impl<C: Send, S: Sync> FromContext<C, S> for Tenant {
    type Rejection = HandlerOutcome;

    fn from_context(
        ctx: &mut Context<'_, C, S>,
    ) -> impl Future<Output = Result<Self, HandlerOutcome>> + Send {
        let tenant = ctx.headers().get("x-tenant");
        let tenant = tenant.map(|tenant| Self(String::from_utf8_lossy(tenant).into_owned()));
        // No tenant, no work: the delivery is dropped and the body does not run.
        ready(tenant.ok_or_else(HandlerOutcome::drop))
    }
}

/// An accepted order and its tenant, published on `accepted`.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Accepted {
    order: u64,
    tenant: String,
}

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

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

impl MessageHeaders for Accepted {
    type Contract = NoHeaders;
}

/// The body resolves the tenant first; a rejection settles the delivery in its place.
struct Accept;

impl Handle<Order, Accepted> for Accept {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> Result<Accepted, HandlerOutcome> {
        let Tenant(tenant) = Tenant::from_context(ctx).await?;
        Ok(Accepted {
            order: order.id,
            tenant,
        })
    }
}