Custom extractors and decoders

When built-in extractors are insufficient, you can derive application values from metadata or write a custom body decoder. These use different traits: FromDeliveryParts does not consume body authority, while FromDelivery owns the single terminal payload.

Write a metadata extractor#

use lilyrs::queue::{
    DeliveryInvocation, EventId, FromDeliveryParts, QueueHandlerError,
};

struct MessageIdentity(EventId);

impl FromDeliveryParts for MessageIdentity {
    type Rejection = QueueHandlerError;

    async fn from_delivery_parts(invocation: &mut DeliveryInvocation)
        -> Result<Self, Self::Rejection>
    {
        Ok(Self(invocation.context().event_id()))
    }
}

MessageIdentity wraps the validated envelope identity in an owned application type. A handler can use MessageIdentity(id): MessageIdentity before its payload parameter. This changes the data representation; it does not maintain a deduplication record by itself.

DeliveryInvocation provides bounded metadata, a read-only DI provider, and typed local state. Parts extractors return owned values, rather than handler parameters borrowing the invocation. Rejections must implement Into<QueueHandlerError> + Send. For optional extraction, implement OptionalFromDeliveryParts and represent genuine absence as Ok(None) without hiding malformed input as absence.

Define a custom body contract#

The complete alternative application below decodes a UTF-8 number with the vendor.order-id content token into an OrderId. Reuse the first application's Cargo manifest and lily.toml, replacing src/main.rs with this file.

use lilyrs::{
    consumer::{Consumer, ConsumerError, Injectable, ServiceTrait},
    queue::{DeliveryPayloadInput, FromDelivery, QueueHandlerError, QueuePayloadKind, queue, queue_service},
};
struct OrderId(u64);
impl FromDelivery for OrderId {
    const PAYLOAD_KIND: QueuePayloadKind = QueuePayloadKind::Custom;
    type Rejection = QueueHandlerError;
    async fn from_delivery(mut input: DeliveryPayloadInput<'_>) -> Result<Self, Self::Rejection> {
        let body = input.take_body()?;
        let text = std::str::from_utf8(&body)
            .map_err(|_| QueueHandlerError::permanent("ORDER_ID_ENCODING_INVALID"))?;
        let id = text.parse::<u64>()
            .map_err(|_| QueueHandlerError::permanent("ORDER_ID_INVALID"))?;
        if id == 0 { return Err(QueueHandlerError::permanent("ORDER_ID_ZERO")); }
        Ok(Self(id))
    }
}
#[derive(Default, Injectable)]
#[service(lifetime = "Singleton")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}
#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "vendor.order-id")]
    async fn created(&self, OrderId(id): OrderId) -> Result<(), QueueHandlerError> {
        println!("{id}");
        Ok(())
    }
}
#[tokio::main]
async fn main() -> Result<(), ConsumerError> { Consumer::run().await }

PAYLOAD_KIND = QueuePayloadKind::Custom declares the extractor's custom body authority to the registry. take_body() consumes the body once from that input. UTF-8, numeric format, and zero checks are explicit application decoder decisions, each producing a permanent error with a safe code.

Use a matching producer#

The first guide's publish_json call does not match this alternative handler. Add the following function to the separate publisher application and call it in place of the existing publish call, preserving container cleanup:

async fn publish_order_id(
    publisher: &lilyrs::queue_client::QueueClientService,
) -> Result<(), Box<dyn std::error::Error>> {
    use lilyrs::queue_client::{CustomPublishContent, PublishSchemaVersion};
    let content = CustomPublishContent::try_new("vendor.order-id", "text/plain")?;
    publisher.publish_custom(
        "orders", "orders.created", PublishSchemaVersion::V1, content, b"42",
    ).await?;
    Ok(())
}

The publisher and handler use the exact same vendor.order-id token. Choosing the text/plain MIME type does not change the content kind to text. The built-in json, text, and binary tokens belong to their corresponding typed publish methods.

Decoder boundaries and documentation#

Custom decoding does not disable transport size bounds or envelope validation. Blocking or lengthy CPU work can occupy the shared async executor; an asynchronous timeout cannot preempt non-yielding code. Design external decoder calls around the deadline and cancellation policy.

Writing a custom decoder does not automatically generate an AsyncAPI schema. When using AsyncAPI, custom or raw payloads need explicit schema or opaque metadata and a content type; explicitly skipping the handler is another option.