Consumer’ı uygulamadan yönetme

start_managed() Consumer’ı başlatır ve clone edilebilen bir ManagedConsumer handle döndürür. Health gözlemi veya üst uygulamanın kapanış koordinasyonu gerektiğinde bu yolu seçin. Standalone process için Consumer::run() daha kısa sinyal tabanlı yoldur; public builder üzerinde ayrıca build() veya spawn() yoktur.

Başlatın ve terminal sonucu bekleyin#

Bu tam örnek doğal runtime termination veya Ctrl+C’yi bekler. Ardından aynı retained task’ın sonucunu shutdown() ile alır; hata yolunda da raporu okur. İlk uygulamanın broker/topology configuration’ını kullanır.

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(())
}

Örneği indirin. Facade consumer ve serde/derive yanında Tokio macros, rt-multi-thread, signal feature’larını açın. Örnek yalnız Ctrl+C listener’ını kendisi ekler; managed mod platform SIGTERM/SIGQUIT listener’larını otomatik kurmaz. Deployment trigger’larını üst uygulamanız bağlamalıdır.

Handle işlemlerinin farkı#

Başarılı startup’tan hemen sonra broker recovery yarışabilir; handle almak kalıcı readiness garantisi değildir. Her health isteğinde snapshot().ready değerini yeniden okuyun. Eşzamanlı veya tekrarlı wait/shutdown çağrıları yeni cleanup denemesi oluşturmaz; aynı shared task ve immutable sonucu gözler.

Dış container kullanımı#

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
}

Container caller-owned kalır; Consumer kendi queue runtime’ını kapatır ama dış container’ı dispose etmez. Üst owner önce Consumer ve diğer kullanıcıların bitmesini bekler, sonra container’ı kapatır. Dış telemetry seçildiyse onun flush/close sorumluluğu da dış owner’dadır. .container(...) ile .secret_resolver(...) birlikte kullanılamaz; config’i dış container’a seed edin.

Drop ve cancellation sınırları#

Bir wait() veya shutdown() observation future’ını bırakmak shared runtime task’ını abort etmez. Son handle clone’u Drop olduğunda graceful shutdown istenir, fakat Rust Drop await edemediği için sonucu gözlemez. Startup readiness waiter’ı handle alınmadan bırakılırsa startup/runtime cancellation tetiklenir.

Cleanup’ın ilerleyebilmesi için Tokio runtime canlı kalmalıdır. Drop veya cancel() çağrısını başarılı cleanup kanıtı saymayın; açık sonucu ve shutdown raporunu bekleyin. Keyfi tokio::spawn application task’ları için Consumer background-service registry’si yoktur; bu task’ların cancel/join sahipliği uygulamadadır.