Managing the Consumer

start_managed() starts the Consumer and returns a cloneable ManagedConsumer handle. Choose it when you need health observations or shutdown coordination by a parent application. Consumer::run() is the shorter signal-based path for a standalone process; the public builder has no additional build() or spawn() method.

Start and observe the terminal result#

This complete example waits for natural runtime termination or Ctrl+C. It then obtains the same retained task’s result through shutdown() and reads the report even on error. It uses the broker/topology configuration from your first application.

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

#[derive(Deserialize)]
struct OrderCreated { order_id: String }
#[derive(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> {
        if event.order_id.is_empty() {
            return Err(QueueHandlerError::permanent("ORDER_ID_REQUIRED"));
        }
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let consumer = Consumer::builder().start_managed().await?;
    println!("ready={}", consumer.snapshot().ready);
    let signal_error = tokio::select! {
        _ = consumer.wait() => None,
        signal = tokio::signal::ctrl_c() => signal.err(),
    };
    // Observe the retained terminal result, including natural runtime failure.
    let result = consumer.shutdown().await;
    if let Some(report) = consumer.shutdown_report() {
        println!("completion={:?}, successful={}", report.completion, report.is_success());
    }
    if let Some(error) = &signal_error { eprintln!("Signal listener failed: {error}"); }
    result?;
    if let Some(error) = signal_error { return Err(error.into()); }
    Ok(())
}

Download the example. Alongside facade consumer and serde/derive, enable Tokio macros, rt-multi-thread and signal. The example installs only its own Ctrl+C listener; managed mode does not automatically install platform SIGTERM/SIGQUIT listeners. Connect deployment triggers in the parent application.

How handle operations differ#

Broker recovery can race immediately after successful startup; receiving the handle does not guarantee permanent readiness. Read snapshot().ready again for every health request. Concurrent or repeated wait/shutdown calls do not create new cleanup attempts; they observe the same shared task and immutable result.

Using an external container#

use std::sync::Arc;
use lilyrs::consumer::{ApplicationContainer, Consumer, ConsumerError, ManagedConsumer};

async fn start_with_shared_container(
    container: Arc<ApplicationContainer>,
) -> Result<ManagedConsumer, ConsumerError> {
    Consumer::builder().container(container).start_managed().await
}

The container remains caller-owned; Consumer closes its queue runtime but does not dispose the external container. The parent owner first waits for Consumer and other users, then closes the container. If telemetry is external, its flush/close belongs to the external owner too. .container(...) cannot be combined with .secret_resolver(...); seed config into the external container.

Drop and cancellation boundaries#

Dropping a wait() or shutdown() observation future does not abort the shared runtime task. Dropping the last handle clone requests graceful shutdown, but Rust Drop cannot await its result. Dropping the startup readiness waiter before receiving a handle signals startup/runtime cancellation.

The Tokio runtime must remain alive for cleanup to progress. Do not treat Drop or a cancel() call as evidence of successful cleanup; await the explicit result and shutdown report. Consumer has no background-service registry for arbitrary application tokio::spawn tasks; the application owns their cancellation and joins.