Middleware

Middleware supplies delivery behavior shared across handlers, such as inspecting metadata, producing typed local values, or observing pipeline results. QueueMiddleware is a concrete type that Consumer constructs once and shares across selected handler plans.

A complete global middleware example#

The following middleware.rs file replaces the first application's src/main.rs. It uses the same Cargo manifest, lily.toml, and JSON publisher. Audit inserts a local marker before execution, and the handler receives it as a typed parameter.

use std::sync::Arc;
use lilyrs::{
    consumer::{Consumer, ConsumerError, Extensions, Injectable, ServiceTrait},
    queue::{
        Json, Local, QueueDeliveryExchange, QueueDeliveryOutcome, QueueHandlerError,
        QueueMiddleware, QueuePipelineComponentInitError, async_trait, queue, queue_service,
    },
};
use serde::Deserialize;

#[derive(Clone)]
struct AuditMarker;

struct Audit;
#[async_trait]
impl QueueMiddleware for Audit {
    async fn new(_: Arc<Extensions>) -> Result<Self, QueuePipelineComponentInitError> {
        Ok(Self)
    }

    async fn before_delivery(&self, exchange: &mut QueueDeliveryExchange<'_>)
        -> Result<(), QueueHandlerError>
    {
        exchange.insert_local(AuditMarker)?;
        Ok(())
    }

    async fn after_delivery(&self, _: &mut QueueDeliveryExchange<'_>, outcome: QueueDeliveryOutcome)
        -> Result<(), QueueHandlerError>
    {
        println!("pipeline_success={}", outcome.is_success());
        Ok(())
    }
}

#[derive(Deserialize)]
struct OrderCreated { order_id: String }

#[derive(Default, Injectable)]
#[service(lifetime = "Singleton")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json")]
    async fn created(&self, Local(_audit): Local<AuditMarker>, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        println!("{}", event.order_id);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), ConsumerError> {
    Consumer::builder().middleware::<Audit>().run().await
}

.middleware::<Audit>() registers global middleware. A middleware is considered entered after before_delivery returns Ok(()). after_delivery runs during normal reverse exit; outcome.is_success() describes the pipeline result, rather than proof of a broker ACK or database commit.

Registration levels and execution order#

before: global → service → handler
guards: global → service → handler
typed extraction → business handler
after: handler → service → global

Every middleware entry runs before guards, regardless of how builder calls are interleaved. Each marker accepts one type; use two attributes for two components. Repeating a type in the same effective plan, or using one concrete type as both middleware and guard, is a startup error.

The example registers Audit globally, so do not also add it as a service or handler marker. To narrow its scope, remove the global registration and use the lilyrs::queue::middleware marker at the chosen level. A service marker belongs below the outer #[queue_service] attribute.

Normal errors and reverse exit#

If before_delivery returns an error, that middleware is not considered entered; only the previously entered prefix unwinds in reverse order. Guard, extractor, and handler errors also pass through the normal after_delivery path of entered middleware. Design each callback's asynchronous side effects to tolerate cancellation.

An after_delivery error can turn a successful inner result into failure, but cannot erase or replace an existing primary failure. Before hooks, guards, handler resolution, extraction, the handler, and normal after hooks share one normal pipeline deadline. After hooks do not receive an independent timeout renewed for each invocation.

Interruption and the termination hook#

An entered middleware can use on_delivery_termination when normal after never started, was interrupted, or panicked. A normal after that returned either Ok or Err is terminal and is not repeated through this hook. Add the following method to the existing QueueMiddleware for Audit impl:

async fn on_delivery_termination(
    &self,
    context: &mut lilyrs::queue::QueueDeliveryTerminationContext<'_>,
) -> Result<(), QueueHandlerError> {
    let _reason = context.reason();
    let _cleanup_deadline = context.deadline();
    let _marker = context.remove_local::<AuditMarker>();
    Ok(())
}

This fragment only releases the local marker. Real asynchronous cleanup must use the context's separate cleanup cancellation and deadline, rather than treating the execution token as cleanup authority. The hook is best-effort, is not retried after interruption, and its error cannot rewrite the pipeline result or settle the delivery.

DI, local state, and constructor budget#

new(Arc<Extensions>) constructs one component instance for the application lifetime. It can retain Singleton dependencies; resolve Scoped or Transient delivery dependencies inside a callback with exchange.service::<T>().await. Keeping a Transient resolved in the constructor in a field extends it to the component's lifetime.

The aggregate pipeline component construction timeout defaults to 30 seconds; .pipeline_initialization_timeout(Duration) accepts a value greater than zero and at most 300 seconds. It is separate from per-message execution timeouts. Local state holds at most 32 distinct types, and inserting the same type again returns the old value.

Do not assume middleware has an additional application shutdown destructor hook; attach asynchronous resources to the lifecycle of an appropriate DI service. Continue with guards for policy rejection, delivery-local state for typed data sharing, and deadlines for timing bounds.