Named veritabanı bağlantıları · Consumer

Factory feature, database_cell ve collection/repository seçimini aynı veritabanı otoritesinde tutun.

Factory modu, aynı backend için isimlendirilmiş bağlantılar tanımlamanızı sağlar. Transactional queue, database_cell alanıyla tam olarak bir bağlantıya bağlanır. Handler’ın iş verisi, migrator ve transaction storage aynı cell’i kullanmalıdır; isim seçimi ikinci bir pool oluşturmaz.

Single feature’ı factory ile değiştirin#

Backend	Tek bağlantı	Factory
PostgreSQL	`consumer-transactional-inbox-postgresql`	`consumer-transactional-inbox-postgresql-factory`
MongoDB	`consumer-transactional-inbox-mongodb`	`consumer-transactional-inbox-mongodb-factory`

Aynı backend için iki sütunu birlikte açmayın. Factory compound feature’ına ek olarak temel postgresql veya mongodb feature’ını açmak single modu da etkinleştirir. Her backend kendi mode seçimini yapabilir; bir binary’de PostgreSQL ve MongoDB bulunması queue’nun backend seçimini otomatikleştirmez.

PostgreSQL factory#

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer-transactional-inbox-postgresql-factory", "config"] }
serde = { version = "1", features = ["derive"] }
uuid = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

[postgresql]
mode = "factory"

[[postgresql.cells]]
name = "primary"
connection_string = "${secret:postgresql.primary_url}"

[postgresql.cells.tls]
mode = "verify-full"

[rabbitmq.topology.queues.transactional_inbox]
backend = "postgresql"
database_cell = "primary"

DB bölümü top-level’dır; queue addition ise ilk uygulamanın orders.created queue tanımına eklenir. ${secret:...} referansı için kendi resolver’ınızı sağlayın. PgFactory::get("primary") seçilen servisi döndürür veya hata verir. Aşağıdaki modül, aynı cell’den migration yapar ve handler’ın SQL yazısını transaction connection’ında çalıştırır.

use lilyrs::consumer::{ApplicationContainer, Injectable, ServiceTrait};
use lilyrs::queue::{Json, PostgresInboxOutboxMigrator, PostgresTransaction,
    PublishContentKind, QueueHandlerError, TransactionalOutboxMessage, queue, queue_service};
use lilyrs::postgresql::{diesel, diesel_async::RunQueryDsl};
use serde::Deserialize;
use uuid::Uuid;

#[derive(Deserialize)]
struct OrderCreated { order_id: i64, outgoing_event_id: String }
#[derive(Default, Injectable)]
#[service(lifetime = "Scoped")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json", delivery_guarantee = "transactional_inbox")]
    async fn created(&self, transaction: PostgresTransaction, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        let outgoing = Uuid::parse_str(&event.outgoing_event_id)
            .map_err(|_| QueueHandlerError::permanent("ORDER_EVENT_ID_INVALID"))?;
        transaction.with_connection(move |connection| Box::pin(async move {
            // Create this application table in your deployment migration first.
            diesel::sql_query("INSERT INTO orders (id) VALUES ($1)")
                .bind::<diesel::sql_types::BigInt, _>(event.order_id)
                .execute(connection).await?;
            Ok(())
        })).await?;
        transaction.enqueue(TransactionalOutboxMessage::try_new(
            outgoing, "domain.events", "orders.persisted", 1,
            PublishContentKind::Json, br#"{"status":"persisted"}"#.as_slice(),
        )?).await?;
        Ok(())
    }
}

// Invoke from a separate deployment migration command, before Consumer startup.
pub async fn migrate(container: &ApplicationContainer) -> Result<(), Box<dyn std::error::Error>> {
    let database = container.resolve::<lilyrs::postgresql::PgFactory>(None).await?.get("primary")?;
    PostgresInboxOutboxMigrator::new(database).migrate().await?;
    Ok(())
}

Tam postgresql_factory.rs örneğini indirin.

MongoDB factory#

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer-transactional-inbox-mongodb-factory", "config"] }
serde = { version = "1", features = ["derive"] }
uuid = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tokio-util = "0.7"

[database]
mode = "factory"

[[database.cells]]
name = "primary"
database_type = "mongodb"
connection_string = "${secret:mongodb.primary_url}"
database_name = "orders"

[rabbitmq.topology.queues.transactional_inbox]
backend = "mongodb"
database_cell = "primary"

MongoFactory::get("primary") bulunamayan cell için None döndürür. Collection derive’ında #[cell_name("primary")] aynı seçimi yapar. Factory modunda mongo_factory inject edilir; derive’ın kullandığı db alanı tutulur fakat single DatabaseService olarak inject edilmez. Aşağıdaki dosya PostgreSQL örneğine alternatiftir; aynı queue contract’ına iki örneği birlikte bağlamayın.

use std::sync::Arc;
use lilyrs::consumer::{ApplicationContainer, Injectable, ServiceTrait};
use lilyrs::mongodb::{Collection, DatabaseService, MongoCollection};
use lilyrs::queue::{Json, MongoInboxOutboxMigrator, MongoTransaction,
    PublishContentKind, QueueHandlerError, TransactionalOutboxMessage, queue, queue_service};
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

#[derive(Clone, Serialize, Deserialize)]
struct Order { order_id: i64 }
#[derive(Default, Injectable, MongoCollection)]
#[service(lifetime = "Singleton")]
#[collection("orders")]
#[collection_type(Order)]
#[cell_name("primary")]
struct Orders {
    db: Arc<DatabaseService>,
    #[inject]
    mongo_factory: Arc<lilyrs::mongodb::MongoFactory>,
    collection: Option<Collection<Order>>,
}
#[derive(Deserialize)]
struct OrderCreated { order_id: i64, outgoing_event_id: String }
#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct OrderWorker {
    #[inject]
    orders: Arc<Orders>,
}
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json", delivery_guarantee = "transactional_inbox")]
    async fn created(&self, transaction: MongoTransaction, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        let outgoing = Uuid::parse_str(&event.outgoing_event_id)
            .map_err(|_| QueueHandlerError::permanent("ORDER_EVENT_ID_INVALID"))?;
        let operation = transaction.operation_context()?;
        self.orders.insert_one(Order { order_id: event.order_id }, &operation).await
            .map_err(|_| QueueHandlerError::retryable("ORDER_WRITE_FAILED"))?;
        transaction.enqueue(TransactionalOutboxMessage::try_new(
            outgoing, "domain.events", "orders.persisted", 1,
            PublishContentKind::Json, br#"{"status":"persisted"}"#.as_slice(),
        )?).await?;
        Ok(())
    }
}

// Apply application collection migrations before the Lily component migration.
pub async fn migrate(container: &ApplicationContainer) -> Result<(), Box<dyn std::error::Error>> {
    let database = container.resolve::<lilyrs::mongodb::MongoFactory>(None).await?
        .get("primary").ok_or("primary MongoDB cell is missing")?;
    let operation = database.operation_context(CancellationToken::new())?;
    MongoInboxOutboxMigrator::new(database.as_ref())?.apply(&operation).await?;
    Ok(())
}

Tam mongodb_factory.rs örneğini indirin.

Factory örnekleri PostgreSQL transaction veya MongoDB transaction örneğinin aynı iş contract’ını kullanır. Önce business migration’ını, sonra seçilen cell’de component migration’ını uygulayın; MongoDB için transaction-capable replica set/sharded deployment gerekir. Örnekteki DB hata eşlemesi sadeleştirilmiştir: kalıcı validation/unique hatalarını uygulamanızın error policy’sinde ayrı sınıflandırın.

Her iki örneğe de numeric order_id ve geçerli non-nil UUID outgoing_event_id içeren payload gönderin; ilk uygulamadaki string order_id payload’ını değiştirmeden kullanmayın. Incoming canonical event ID, bu outgoing ID’den ayrıdır. Outgoing domain.events exchange’i ile orders.persisted routing key’inin destination binding’ini ayrıca provision edin.

{"order_id": 42, "outgoing_event_id": "e7b04866-8f2e-46e4-a92c-d35dd2016879"}

Manifestler config ve ilk uygulamanın Tokio runtime feature’larını içerir. Secret reference’ları çözmek için secrets rehberindeki resolver’ı kurun. PostgreSQL factory cell TLS’i verify-full kullanır; gerekli trust/hostname ayarlarını PostgreSQL rehberinden tamamlayın.

Cell sınırları ve paylaşım#

Factory modunda database_cell zorunludur; single modunda yasaktır. Lily ilk cell’i seçmez ve farklı isim yazımını düzeltmez. Migration’ı bir cell’e uygulayıp queue’yu diğerine bağlarsanız ilgili storage preflight başarısız olur. Mongo collection’ının farklı cell’i seçmesi de transaction session otoritesiyle uyumsuzdur.

Aynı backend/cell’e bağlı queue’lar bir outbox relay paylaşır ve bütün transactional policy değerleri eşit olmalıdır. Bağlantı yaşam döngüsü DI owner’ındadır; queue transaction binding’i ek bir container veya client üretmez. Bu örneklerdeki migrate fonksiyonları ayrı deployment komutunun body’sidir; migration rehberindeki bootstrap ve shutdown sırası yine geçerlidir.