Your first Consumer application

This page builds an application that reads JSON messages from orders.created and prints the order ID. The next publisher guide sends an order-42 message using the same contract.

Create the project#

Use Rust 1.96.1 and edition 2024. Create an independent Cargo project and replace Cargo.toml with the manifest below. The published lilyrs 0.1.0 registry package is sufficient; the framework repository is not a required dependency checkout.

cargo new orders-consumer
cd orders-consumer

[package]
name = "orders-consumer"
version = "0.1.0"
edition = "2024"
rust-version = "1.96.1"

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

consumer also enables queue types and DI support. serde decodes the JSON message. The tokio dependency provides the #[tokio::main] runtime; Consumer does not create a new Tokio runtime itself.

Write a worker and handler#

Download the complete file below and save it as src/main.rs:

use lilyrs::{
    consumer::{Consumer, ConsumerError, 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(())
    }
}

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

OrderWorker registers with DI as a Singleton. Inside #[queue_service], #[queue] declares the physical queue name and accepted version/content pair. Json<OrderCreated> decodes the body; Ok(()) reports normal pipeline success. Workers are discovered at link time, without a manually maintained add_handler list.

Configure RabbitMQ and the queue#

Your local RabbitMQ instance must be available at 127.0.0.1:5672 and allow the guest user to access the / vhost. This is a local development connection; the example does not install a broker. Download the configuration below and place it at the project root as lily.toml:

[rabbitmq.consumer]
hostname = "127.0.0.1"
port = 5672
username = "guest"
password = "guest"
vhost = "/"
use_tls = false
pool_size = 1

[[rabbitmq.topology.queues]]
name = "orders.created"
exchange_name = "orders"
routing_key = "orders.created"
concurrency = 2
prefetch_count = 10
retry_attempts = 3
retry_backoff_millis = 1000
max_retry_backoff_millis = 60000

[rabbitmq.topology.queues.retention]
main_max_messages = 100000
main_max_bytes = 1073741824
retry_bucket_max_messages = 10000
retry_bucket_max_bytes = 268435456
dead_letter_max_messages = 10000
dead_letter_max_bytes = 268435456

The connection section says how to reach the broker; topology defines the exchange and routing key used by orders.created. All six retention values are required. They bound message count and bytes for the main queue, retry buckets, and dead-letter queue. These example values should be adjusted to your application's capacity.

concurrency = 2 bounds concurrently executing deliveries; prefetch_count = 10 is the broker's unacknowledged delivery window. These settings control different things. See capacity and topology for details.

Run the application and send a message#

LILY_CONFIG_PATH=./lily.toml LILY_CONFIG_MODE=development cargo run

Wait for the Consumer to connect and complete queue registration. The broker management interface can show an active consumer for orders.created; use managed execution to query readiness from your application. Then run the publisher example in another terminal. This terminal prints order-42.

cargo check checks the Rust contract. Broker access and end-to-end delivery require running both programs. Tracing is disabled by default, and Consumer opens no HTTP port. An automatically provided /health endpoint is therefore not part of startup verification.

Shut down the application#

Ctrl+C stops new admission and starts reconciling accepted work within the shutdown budget. Await the result of Consumer::run().await. The default application shutdown budget is 30 seconds. See shutdown for interpreting the result and distinguishing incomplete cleanup.