Queue services and handlers

A queue service is a DI service containing one or more message handlers. DI determines the service lifetime, #[queue] declares the message contract, and configuration defines the physical queue's runtime policy.

Register the worker service#

use lilyrs::{
    consumer::{Injectable, ServiceTrait},
    queue::{Json, QueueHandlerError, queue, queue_service},
};
use serde::Deserialize;

#[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, Json(event): Json<OrderCreated>) -> Result<(), QueueHandlerError> {
        println!("{}", event.order_id);
        Ok(())
    }
}

This fragment is the worker portion of the first application; its process entry calls Consumer::run().await. Injectable creates the DI registration, while queue_service supplies handler metadata. When moving a worker to another module, include that module in the compiled Rust module tree.

Macro and signature rules#

&mut self, generic handlers, reference extractors, and non-unit success values do not satisfy this contract. #[queue] is not placed on a free function. concurrency, retry, and retention are configuration settings, rather than macro arguments; see configuration.

Multiple contracts on one queue#

(orders.created, 1, json)   → OrderWorker::created_v1
(orders.created, 2, json)   → OrderWorker::created_v2
(orders.created, 2, binary) → OrderWorker::created_binary_v2

Every queue/version/content tuple must be unique. This example has one physical orders.created queue entry in TOML, with three handlers selected by message contract. Registering the same tuple in another worker causes a startup error; an ambiguous registration remains invalid even when its queue is inactive in the selected configuration.

A valid linked handler whose queue is absent from configuration does not create active consumption. Conversely, a configured physical queue with no handler is an error. Handler selection does not make the broker route headers to different processes; rollout order must account for the versions each replica accepts.

Service lifetime and pipeline#

A Singleton worker can retain shared, thread-safe application state. A Scoped worker resolves within the admitted delivery's scope after middleware and guards. Singleton ownership is not mandatory, but a DI registration is required. Use the Service extractor to obtain business services in handlers.

Service and handler levels can use #[middleware(Type)] and #[guard(Type)] markers. Each marker accepts one type and is processed by the outer #[queue_service]. See middleware and guards for registration order.