İlk mesajı yayınlayın

Consumer'ın mesaj işlediğini görmek için aynı sözleşmeyi kullanan bir producer gerekir. Bu sayfada ayrı bir Cargo uygulaması orders exchange'ine, orders.created routing key'iyle JSON yayımlar.

Önce topology hazır olsun#

İlk Consumer uygulamasını önce başlatın ve queue registration'ın tamamlanmasını bekleyin. QueueClientService initialization veya publish_json exchange, queue ya da binding oluşturmaz. Publisher'ın publish yaptığı anda orders exchange'i ve orders.created binding'i var olmalıdır.

Topology'yi publisher tarafından kurmanız gereken bir deployment'ta açık RabbitMqTopologyBootstrap::from_client(...).run(...) işletim adımı kullanılır; named publisher için from_cell vardır. Bu ayrı kararı topology bölümünde ele alıyoruz. Buradaki akış topology sahipliğini Consumer'da tutar.

Publisher projesini hazırlayın#

cargo new orders-publisher
cd orders-publisher

[package]
name = "orders-publisher"
version = "0.1.0"
edition = "2024"
rust-version = "1.96.1"

[dependencies]
lilyrs = { version = "0.1.0", features = ["queue-client", "injection"] }
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

Bu process Consumer çalıştırmadığı için queue-client ve container'ı kendisi kuracağı için injection feature'ını kullanır. Consumer projesinin Cargo dosyasını değiştirmeyin; iki uygulamanın profilleri ayrıdır.

Aşağıdaki publisher.toml dosyasını indirin ve publisher projesinin köküne koyun. Consumer'ın lily.toml dosyasından ayrıdır:

[queue_client]
mode = "single"
hostname = "127.0.0.1"
port = 5672
username = "guest"
password = "guest"
vhost = "/"
use_tls = false
pool_size = 1
persistence_enabled = true

Mesajı yayınlayın#

Tam publisher dosyasını src/main.rs olarak kaydedin:

use lilyrs::{
    injection::ApplicationContainer,
    queue_client::{PublishSchemaVersion, QueueClientService},
};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let container = ApplicationContainer::build().await?;
    let publish = async {
        let publisher = container.resolve::<QueueClientService>(None).await?;
        publisher.publish_json(
            "orders", "orders.created", PublishSchemaVersion::V1,
            serde_json::json!({"order_id": "order-42"}),
        ).await?;
        Ok::<(), Box<dyn std::error::Error>>(())
    }.await;
    let cleanup = container.close().await;
    // Always await container cleanup, including after publish failure.
    if let Err(error) = &cleanup { eprintln!("Publisher cleanup failed: {error}"); }
    publish?;
    cleanup?;
    Ok(())
}

PublishSchemaVersion::V1 uygulama mesajının sürümüdür. publish_json body encoding'ini, json content-kind'ını ve JSON MIME type'ını birlikte belirler; yeni canonical event ID üretir. İlk handler aynı queue/version/content sözleşmesini kabul eder.

LILY_CONFIG_PATH=./publisher.toml LILY_CONFIG_MODE=development cargo run

Başarılı akışta Consumer terminali order-42 yazdırır. Publisher'ın Ok sonucu, mandatory mesajın unroutable olarak dönmediğini ve broker confirm ACK alındığını gösterir. Handler'ın tamamlandığı veya bir veritabanı transaction'ının commit edildiği anlamına gelmez.

Kaynakları ve tekrar gönderimi yönetin#

Örnek, publish hatası olsa da container kapanışını bekler ve cleanup hatasını görünür kılar. Bir HTTP veya WebSocket handler'ından publisher resolve ediyorsanız host'un container'ını o handler içinde kapatmayın; container sahibi host'tur.

Yeni iş olayı için yeni event ID uygundur. Aynı olayı tekrar gönderirken saklanmış kimliği PublishMetadata::try_new(stored_id, version) ile kullanın ve metadata kabul eden publish metodunu seçin. Her denemede yeni kimlik üretmek deduplication zincirini koparır. Durable DB publication için transactional outbox ayrı bir model sunar.