Arka plan servisleri

Bağlantıdan veya tek bir mesajdan bağımsız periyodik bir iş için BackgroundServiceTrait kullanın. WsAppBuilder::add_background_service::<T>() worker'ı uygulamanın yaşam döngüsüne bağlar: hazırlama, başlatma, cancellation ve kapanışı bekleme sorumluluğunu Lily üstlenir.

Bağımlılıklar ve kayıt#

İlk uygulamanın lilyrs bağımlılığı yeterlidir; ayrıca background-service veya injection feature'ı gerekmez. Örnekte zamanlayıcı kullanıldığı için Tokio satırını şöyle yazabilirsiniz:

lilyrs = { version = "0.1.0", features = ["websocket"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }

Aşağıdaki dosyayı src/background.rs olarak kaydedin ve crate köküne mod background; ekleyin. İlk uygulamanın builder zincirinde, build öncesinde .add_background_service::<background::ProcessingWorker>() çağırın. İsterseniz dosyanın with_worker(builder) fonksiyonu aynı kaydı yapar. Controller'ların link-time keşfi bu kayıttan bağımsızdır.

Her iş için scope açın#

use std::{sync::Arc, time::Duration};
use lilyrs::websocket::{
    ApplicationScopeFactory, BackgroundCancellation, BackgroundServiceTrait,
    Injectable, InjectionError, ProcessContext, ServiceTrait, WsAppBuilder, async_trait,
};

#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct ProcessingService;
impl ServiceTrait for ProcessingService {}
impl ProcessingService {
    async fn process_next(&self, stopping: BackgroundCancellation) -> Result<(), InjectionError> {
        if stopping.is_cancelled() { return Ok(()); }
        // Replace this with one bounded application operation.
        tokio::task::yield_now().await;
        Ok(())
    }
}

pub struct ProcessingWorker {
    scopes: Arc<ApplicationScopeFactory>,
}

#[async_trait]
impl BackgroundServiceTrait for ProcessingWorker {
    type Error = InjectionError;

    async fn new(scopes: Arc<ApplicationScopeFactory>) -> Result<Self, Self::Error> {
        Ok(Self { scopes })
    }

    async fn execute_async(&mut self, stopping: BackgroundCancellation) -> Result<(), Self::Error> {
        while !stopping.is_cancelled() {
            let signal = stopping.clone();
            self.scopes.create_scope(ProcessContext::new())?
                .run(move |extensions| Box::pin(async move {
                    extensions.get_service::<ProcessingService>(None).await?
                        .process_next(signal).await
                }))
                .await?;
            tokio::select! {
                _ = stopping.cancelled() => break,
                _ = tokio::time::sleep(Duration::from_secs(1)) => {},
            }
        }
        Ok(())
    }
}

pub fn with_worker(builder: WsAppBuilder) -> WsAppBuilder {
    builder.add_background_service::<ProcessingWorker>()
}

Tam örnek dosyasını indirin.

process_next içindeki örnek gövde iş yükünü temsil eder; gerçek uygulamada buraya bounded veri erişimi veya tek bir iş birimi yerleştirin. Worker, Scoped ProcessingService instance'ını alanında tutmaz. ApplicationScopeFactory ile her tur için scope açar; ServiceScope::run doğru ProcessContext içinde servisi çözer ve scope disposal'ını bekler. Callback'teki servis veya Extensions değerini scope dışına taşımayın, bunlarla detached task açmayın.

Ne zaman çalışır?#

new build sırasında çağrılır. execute_async, listener bind ve varsa Required backplane subscription hazır olduktan sonra bir kez başlatılır; Optional backplane readiness'i beklenmez. Başarısız startup veya start öncesi close worker'ı execute etmez. Uygulamanın readiness'i worker'ın ilk işinin tamamlanmasını beklemez; bir dependency için böyle bir kural gerekiyorsa health kaydını uygulamanız kurar.

Aynı concrete worker tipi uygulama başına bir kez kaydedilir. Aynı tipi ayrı HTTP ve WebSocket uygulamalarına eklemek iki bağımsız worker oluşturur. Ok(()) tek seferlik veya devre dışı bir worker için normal bitiştir. İşlenmeyen hata ya da panic host kapanışını başlatır; otomatik retry veya restart yoktur. Beklenen shutdown cancellation'ını işleyip Ok(()) döndürün; domain retry politikasını bounded biçimde worker içinde kurun.

İki cancellation tipi#

BackgroundCancellation, ortak lily_cancellation::ExecutionCancellation tipinin alias'ıdır. Mesaj action/middleware'larının kullandığı lilyrs::websocket::ExecutionCancellation farklıdır. Bir mesajın timeout'u veya peer disconnect worker'ı iptal etmez; worker sinyali host shutdown'a bağlıdır. Mesaj cancellation'ını worker parametresi yerine kullanmayın.

İşlemler kendi bounded sürelerini ve cancellation sözleşmelerini korumalıdır. Cancellation tek başına veritabanı commit'ini geri almaz. Shutdown'a yanıt verirken finalization için yeni scope açılabilir; worker'lar sona erdiğinde veya forced stopping başladığında yeni scope kabulü kapanır.

Kapanış ve sahiplik#

Worker'lar uygulamanın ortak [lifecycle].shutdown_timeout_secs bütçesini kullanır. Lily cancellation'dan sonra bounded cooperation ve gerçek task join'lerini, scope cleanup'ını, backplane/owned DI kapanışını ve telemetry flush'ını izler. Ayrı bir worker timeout'u ile her aşamada yeni bütçe açılmaz. Waiter'ı düşürmek bu sahipleri ortadan kaldırmaz.

Senkron bloklayan future poll veya destructor zorla sonlandırılamaz. Sonlanma doğrulanamazsa kapanış eksik olarak sonuçlanır ve kullanılan bağımlılıklar korunur. Dış container kendi sahibinde kalır. Shutdown rehberi sonucu nasıl yorumlayacağınızı açıklar.

Worker'ın kapanışa yardımcı olması ona aktif WebSocket callback'inin private outbound continuation yetkisini vermez. Shutdown sırasında dispatcher yeni dış send'leri reddeder; kapanan peer'a son bildirim ulaştırılacağını varsaymayın. tokio::spawn ile ayrıca açtığınız task'lar bu worker kaydı sayesinde kendiliğinden managed olmaz.

Gözlem ve test#

Worker lifecycle olayları uygulamanın tracing runtime'ını kullanır; kapanışın flush öncesi checkpoint'i worker/scope sayaçlarını içerir. Tracing ve health üzerinden kendi iş sonucunu ve readiness ihtiyacını ayrı gözlemleyin.

Testlerinizde başarılı startup, start öncesi close, iş başına scope paylaşımı/cleanup, cooperative cancellation ve işlenmeyen worker hatasının host'u kapatmasını doğrulayın. Scope disposal veya task join sonucunu yalnız cancellation sinyalinin gönderilmiş olmasından çıkarmayın. Canlı dış sistem işlemleri ayrıca entegrasyon testi gerektirir.