Özel extractor ve decoder

Built-in extractor'lar ihtiyacınızı karşılamadığında metadata'dan uygulama değeri çıkarabilir veya body için özel decoder yazabilirsiniz. İki işin trait'i ayrıdır: FromDeliveryParts body authority'yi tüketmez, FromDelivery tek terminal payload sahibidir.

Metadata extractor yazın#

use lilyrs::queue::{
    DeliveryInvocation, EventId, FromDeliveryParts, QueueHandlerError,
};

struct MessageIdentity(EventId);

impl FromDeliveryParts for MessageIdentity {
    type Rejection = QueueHandlerError;

    async fn from_delivery_parts(invocation: &mut DeliveryInvocation)
        -> Result<Self, Self::Rejection>
    {
        Ok(Self(invocation.context().event_id()))
    }
}

MessageIdentity, doğrulanmış envelope kimliğini owned uygulama tipine taşır. Handler'da payload'dan önce MessageIdentity(id): MessageIdentity parametresi kullanılabilir. Bu dönüşüm veri biçimini değiştirir; kendiliğinden deduplication kaydı tutmaz.

DeliveryInvocation bounded metadata, read-only DI provider ve typed local state sunar. Parts extractor owned değer döndürür; invocation'a referans taşıyan handler parametresi üretmez. Ret tipi Into<QueueHandlerError> + Send olmalıdır. Optional extraction gerekiyorsa OptionalFromDeliveryParts ile gerçek yokluğu Ok(None) olarak modelleyin; bozuk veriyi yokluk gibi gizlemeyin.

Özel body sözleşmesi tanımlayın#

Aşağıdaki tam alternatif uygulama, vendor.order-id content token'ıyla gelen UTF-8 sayıyı OrderId değerine çevirir. İlk uygulamanın Cargo manifesti ve lily.toml dosyası kullanılabilir; src/main.rs yerine bu dosyayı koyun.

use lilyrs::{
    consumer::{Consumer, ConsumerError, Injectable, ServiceTrait},
    queue::{DeliveryPayloadInput, FromDelivery, QueueHandlerError, QueuePayloadKind, queue, queue_service},
};
struct OrderId(u64);
impl FromDelivery for OrderId {
    const PAYLOAD_KIND: QueuePayloadKind = QueuePayloadKind::Custom;
    type Rejection = QueueHandlerError;
    async fn from_delivery(mut input: DeliveryPayloadInput<'_>) -> Result<Self, Self::Rejection> {
        let body = input.take_body()?;
        let text = std::str::from_utf8(&body)
            .map_err(|_| QueueHandlerError::permanent("ORDER_ID_ENCODING_INVALID"))?;
        let id = text.parse::<u64>()
            .map_err(|_| QueueHandlerError::permanent("ORDER_ID_INVALID"))?;
        if id == 0 { return Err(QueueHandlerError::permanent("ORDER_ID_ZERO")); }
        Ok(Self(id))
    }
}
#[derive(Default, Injectable)]
#[service(lifetime = "Singleton")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}
#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "vendor.order-id")]
    async fn created(&self, OrderId(id): OrderId) -> Result<(), QueueHandlerError> {
        println!("{id}");
        Ok(())
    }
}
#[tokio::main]
async fn main() -> Result<(), ConsumerError> { Consumer::run().await }

PAYLOAD_KIND = QueuePayloadKind::Custom, registry'ye bu extractor'ın custom body authority olduğunu bildirir. take_body() aynı input üzerinde body'yi bir kez alır. UTF-8, sayı biçimi ve sıfır kontrolü uygulama decoder'ının açık ret kararlarıdır; her biri safe code taşıyan permanent hata üretir.

Eşleşen producer kullanın#

İlk sayfadaki publish_json bu alternatif handler ile eşleşmez. Ayrı publisher uygulamasına aşağıdaki fonksiyonu ekleyip mevcut publish çağrısı yerine çağırın; container cleanup akışını koruyun:

async fn publish_order_id(
    publisher: &lilyrs::queue_client::QueueClientService,
) -> Result<(), Box<dyn std::error::Error>> {
    use lilyrs::queue_client::{CustomPublishContent, PublishSchemaVersion};
    let content = CustomPublishContent::try_new("vendor.order-id", "text/plain")?;
    publisher.publish_custom(
        "orders", "orders.created", PublishSchemaVersion::V1, content, b"42",
    ).await?;
    Ok(())
}

Publisher vendor.order-id token'ını, handler da aynı exact token'ı kullanır. Burada text/plain MIME seçilmesi content-kind'ı text yapmaz. Built-in json, text, binary token'ları kendi typed publish metotlarına ayrılmıştır.

Decoder sınırları ve belge üretimi#

Custom decoder transport boyut sınırlarını veya envelope doğrulamasını devre dışı bırakmaz. Blocking veya uzun CPU işi shared async executor'ı tutabilir; bir async timeout non-yielding işi preempt edemez. Decoder'daki dış çağrıları deadline ve cancellation politikasıyla tasarlayın.

Özel decoder yazmak AsyncAPI şemasını otomatik üretmez. AsyncAPI seçildiğinde custom/raw payload için açık schema veya opaque metadata'sı ve content type gerekir; handler'ı açıkça skip etmek de bir tercihtir.