Middleware

Middleware, handler'lar arasında ortak delivery davranışı kurar: metadata incelemek, typed local değer üretmek veya pipeline sonucunu gözlemek gibi. QueueMiddleware bir concrete tiptir; Consumer onu bir kez oluşturur ve seçilen handler planlarında paylaşır.

Global middleware ile tam örnek#

Aşağıdaki middleware.rs dosyası, ilk uygulamanın src/main.rs dosyasının alternatifidir. Aynı Cargo manifestini, lily.toml ve JSON publisher'ı kullanır. Audit before aşamasında local marker bırakır; handler bu marker'ı typed parametre olarak alır.

use std::sync::Arc;
use lilyrs::{
    consumer::{Consumer, ConsumerError, Extensions, Injectable, ServiceTrait},
    queue::{
        Json, Local, QueueDeliveryExchange, QueueDeliveryOutcome, QueueHandlerError,
        QueueMiddleware, QueuePipelineComponentInitError, async_trait, queue, queue_service,
    },
};
use serde::Deserialize;

#[derive(Clone)]
struct AuditMarker;

struct Audit;
#[async_trait]
impl QueueMiddleware for Audit {
    async fn new(_: Arc<Extensions>) -> Result<Self, QueuePipelineComponentInitError> {
        Ok(Self)
    }

    async fn before_delivery(&self, exchange: &mut QueueDeliveryExchange<'_>)
        -> Result<(), QueueHandlerError>
    {
        exchange.insert_local(AuditMarker)?;
        Ok(())
    }

    async fn after_delivery(&self, _: &mut QueueDeliveryExchange<'_>, outcome: QueueDeliveryOutcome)
        -> Result<(), QueueHandlerError>
    {
        println!("pipeline_success={}", outcome.is_success());
        Ok(())
    }
}

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

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

.middleware::<Audit>() global kayıttır. before_delivery Ok(()) döndürdükten sonra middleware girilmiş kabul edilir. after_delivery normal ters çıkışta çağrılır; outcome.is_success() pipeline sonucunu gösterir, broker ACK veya DB commit kanıtı değildir.

Kayıt seviyeleri ve çalışma sırası#

before: global → service → handler
guards: global → service → handler
typed extraction → business handler
after: handler → service → global

Bütün middleware girişleri guard'lardan önce çalışır; builder çağrılarının birbirine karışması bu faz sırasını değiştirmez. Her marker tek type alır; iki component için iki attribute yazılır. Aynı effective plan'da aynı type'ı tekrar eklemek veya aynı concrete type'ı hem middleware hem guard yapmak startup hatasıdır.

Örnekte Audit global kayıtlıdır; ona ayrıca service veya handler marker'ı eklemeyin. Kapsamını daraltmak isterseniz global kaydı kaldırıp lilyrs::queue::middleware marker'ını ilgili seviyede kullanın. Service marker'ı en dış #[queue_service] attribute'unun altında yer alır.

Normal hata ve ters çıkış#

Bir before_delivery hata döndürürse o middleware girilmiş sayılmaz; yalnız daha önce başarıyla girilmiş prefix ters sırayla unwind edilir. Guard, extractor veya handler hatası da girilmiş middleware'lerin normal after_delivery yolundan geçer. Callback'in kendi async yan etkilerini cancellation-safe tasarlayın.

after_delivery hatası başarılı inner sonucu hataya çevirebilir; mevcut primary hatayı silmez veya değiştirmez. Before, guard, handler çözümü, extraction, handler ve normal after callback'leri aynı normal pipeline deadline'ını paylaşır. After aşaması için bağımsız, her seferinde yenilenen bir timeout verilmez.

Yarıda kesilme ve termination hook#

Normal after hiç başlamadıysa, kesildiyse veya panic ile çıktıysa girilmiş middleware için on_delivery_termination cleanup yolu kullanılabilir. Ok veya Err ile tamamlanmış normal after terminaldir ve bu hook'ta tekrar çalıştırılmaz. Aşağıdaki method mevcut QueueMiddleware for Audit impl'ine eklenebilir:

async fn on_delivery_termination(
    &self,
    context: &mut lilyrs::queue::QueueDeliveryTerminationContext<'_>,
) -> Result<(), QueueHandlerError> {
    let _reason = context.reason();
    let _cleanup_deadline = context.deadline();
    let _marker = context.remove_local::<AuditMarker>();
    Ok(())
}

Bu parça yalnız local marker'ı bırakır. Gerçek async cleanup, context'in ayrı cleanup cancellation ve deadline sınırını kullanmalıdır; execution token'ı cleanup yetkisi yerine geçirilmez. Hook best-effort'tur, yarıda kesilirse tekrar denenmez ve dönüş hatası pipeline sonucunu değiştiremez veya delivery'yi settle edemez.

DI, local state ve constructor bütçesi#

new(Arc<Extensions>) application lifetime boyunca tek component instance kurar. Singleton bağımlılık burada tutulabilir; Scoped veya Transient delivery bağımlılığını callback içinde exchange.service::<T>().await ile çözün. Constructor'da çözülen Transient değeri field'da tutmak onu component ömrüne uzatır.

Pipeline component construction için toplam timeout varsayılan 30 saniyedir; .pipeline_initialization_timeout(Duration) sıfırdan büyük, en çok 300 saniyelik bir değer kabul eder. Bu, her mesajın execution timeout'ı değildir. Local state en çok 32 distinct type tutar; aynı type tekrar insert edilirse eski değer döner.

Middleware için ayrıca bir application shutdown destructor hook'u varsaymayın; async kaynakları uygun DI servisinin lifecycle'ına bağlayın. Politika ret kararları guard, typed veri paylaşımı delivery-local state, süre sınırları deadline bölümünde devam eder.