把事件中继到另一个 Broker

本指南在与事件所描述的数据相同的事务中写入事件，然后把它发布到另一个 Broker。

一个服务保存订单，然后向 Kafka 发布“订单已下”，这里有两次可能脱节的写入。订单可能已经提交而发布失败，也可能已经发出发布而订单回滚了。事务性 outbox 把它们合在一起：事件是在订单事务中写入的一行，再由中继发布已提交的行。在本 crate 中，事件表是一张队列表，中继是一个普通的订阅。

示例是一个网店。结账请求下单并写入它的事件；中继把每个事件发布到一条总线上，这里是内存 Broker。

订单及其事件

src/main.rs

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, total_cents INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS checkout_jobs (
    id INTEGER PRIMARY KEY,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS order_events (
    id INTEGER PRIMARY KEY,
    aggregate TEXT NOT NULL,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";

宏

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `order_events`: an event waiting for the relay, keyed by its order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "order_events")]
struct OrderEvent {
    #[field(id, generated)]
    id: i64,
    #[field(partition_key)]
    aggregate: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The order and its event, in the checkout's transaction: both commit, or neither does.
async fn place(
    conn: &mut SqliteConnection,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO orders (id, total_cents) VALUES (?, ?)")
        .bind(order.id)
        .bind(order.total_cents)
        .execute(&mut *conn)
        .await?;
    let placed = OrderPlaced {
        order: order.id,
        total_cents: order.total_cents,
    };
    let event = OrderEvent {
        id: 0,
        aggregate: order.id.to_string(),
        locked_until: None,
        payload: serde_json::to_vec(&placed)?,
    };
    event.insert(conn).await?;
    Ok(())
}

#[subscriber(InboxQueue::<Checkout>::new("checkouts"))]
async fn checkout(order: &Order, Ctx(mut tx): Ctx<keys::Tx<Sqlite>>) -> HandlerOutcome {
    match place(&mut tx, order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(subscription = "checkouts", order.id, %error, "order not placed");
            HandlerOutcome::drop()
        }
    }
}

手写

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `order_events`: an event waiting for the relay, keyed by its order.
#[derive(sqlx::FromRow)]
struct OrderEvent {
    id: i64,
    aggregate: String,
    payload: Vec<u8>,
}

impl InboxTable for OrderEvent {
    type Id = i64;
    type Table = InboxSpec<(Key, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("order_events", Column::new("id").generated())
        .partition_key(Column::new("aggregate"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl KeyRow for OrderEvent {
    type Key = String;

    fn partition_key(&self) -> &String {
        &self.aggregate
    }
}

impl PayloadRow for OrderEvent {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

// INSERT INTO `order_events` (`aggregate`, `locked_until`, `payload`) VALUES (?, ?, ?)
const INSERT: Sql<160> = insert::sqlite(&OrderEvent::TABLE.spec());

impl Insert<SqliteConnection> for OrderEvent {
    async fn insert(&self, conn: &mut SqliteConnection) -> Result<(), sqlx::Error> {
        sqlx::query(INSERT.as_str())
            .bind(&self.aggregate)
            .bind(None::<DateTime<Utc>>)
            .bind(&self.payload)
            .execute(conn)
            .await?;
        Ok(())
    }
}

/// The order and its event, in the checkout's transaction: both commit, or neither does.
async fn place(
    conn: &mut SqliteConnection,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO orders (id, total_cents) VALUES (?, ?)")
        .bind(order.id)
        .bind(order.total_cents)
        .execute(&mut *conn)
        .await?;
    let placed = OrderPlaced {
        order: order.id,
        total_cents: order.total_cents,
    };
    let event = OrderEvent {
        id: 0,
        aggregate: order.id.to_string(),
        payload: serde_json::to_vec(&placed)?,
    };
    event.insert(conn).await?;
    Ok(())
}

struct PlaceOrder;

impl Handle<Order, (), (), TxContext<Sqlite>> for PlaceOrder {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, TxContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(mut tx) = Ctx::<keys::Tx<Sqlite>>::from_context(ctx).await?;
        place(&mut tx, order).await.map_err(|error| {
            tracing::warn!(subscription = "checkouts", order.id, %error, "order not placed");
            HandlerOutcome::drop()
        })
    }
}

checkout 以事务模式运行：它通过投递的事务写入订单和事件。它的确认把两者与结账行的结算一起提交。插入失败的订单会回滚它的事件，不为它发布任何东西。处理器之外的代码，例如一个 HTTP 端点，在它自己的 sqlx 事务中写入同样的两行，做法与任务入队相同。这时事件行同样不唤醒任何订阅：中继在它的轮询间隔内发现它。

中继

宏

src/main.rs

use ruststream::memory::{MemoryBroker, MemoryPublish};

/// Each event goes out on the bus as it was written; the acknowledgement deletes its row.
#[subscriber(InboxQueue::<OrderEvent>::new("order-events"), reply)]
async fn relay(event: &OrderPlaced) -> OrderPlaced {
    event.clone()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // The target broker is wrapped first, so its publish token exists before the scopes.
    let bus = MemoryBroker::new().bindable();
    let to_bus = bus.bind(MemoryPublish);
    RustStream::new(AppInfo::new("shop", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Checkout>("checkouts"), |b| {
            b.include(checkout.transactional());
            b.include(relay.workers_by_key(nonzero!(4)))
                .out_reply(to_bus);
        })
        .with_broker(bus, |_b| {})
}

手写

src/main.rs

use ruststream::memory::{MemoryBroker, MemoryPublish};

/// Each event goes out on the bus as it was written; the acknowledgement deletes its row.
struct Relay;

impl Handle<OrderPlaced, OrderPlaced> for Relay {
    fn handle(
        &self,
        event: &OrderPlaced,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<OrderPlaced, HandlerOutcome>> {
        ready(Ok(event.clone()))
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // The target broker is wrapped first, so its publish token exists before the scopes.
    let bus = MemoryBroker::new().bindable();
    let to_bus = bus.bind(MemoryPublish);
    RustStream::new(AppInfo::new("shop", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Checkout>("checkouts"), |b| {
            let checkout = subscriber("checkouts", PlaceOrder)
                .map_source(|_| InboxQueue::<Checkout>::new("checkouts"))
                .transactional();
            b.include(checkout.build());
            let relay = subscriber("order-events", Relay)
                .map_source(|_| InboxQueue::<OrderEvent>::new("order-events"))
                .workers_by_key(nonzero!(4))
                .reply();
            b.include(relay.build()).out_reply(to_bus);
        })
        .with_broker(bus, |_b| {})
}

relay 读取 order_events，以每个事件本身作为对它的应答；应答类型声明了它的目的地 orders。应答经由总线发出：bindable() 包装目标 Broker，bind(MemoryPublish) 为它生成一个发布令牌，.out_reply(to_bus) 通过这个令牌发送 relay 的应答。Kafka 或 NATS 服务在同样的两行中换上自己的 Broker 和发布策略。应答在事件行结算之前发布，确认删除该行。

中继是至少一次的。发布失败的应答会把事件放回队列，在发布之后、确认之前发生的崩溃会再次发布它。不能重复执行的消费者要保存事件的标识，这里是订单，并跳过已经见过的事件。

按聚合保持顺序

aggregate 是事件的分区键，即订单的 id。workers_by_key(4) 同时中继四个订单，并让同一订单的事件保持写入时的顺序。这个顺序在一个中继订阅之内成立。在多个实例上运行、事件必须在所有实例之间按聚合保持顺序的中继，把事件放进 FIFO 组，或者采用咨询锁方式。

中继不需要“已发送”标志，也不需要自己的轮询：领取由队列负责，发布失败后的延迟是一次重试，一个上限加一个死信组则搁置目标一直拒绝的事件。跟踪你发布的消息是另一种形态：outbox 插件记录处理器发布的内容，并在消费者处理之后标记它，但它的记录单独提交，在业务事务之外。

测试

示例文件中的测试通过测试框架发布结账请求，并用 tb.broker::<MemoryBroker>().published(..) 读取到达总线的内容。一个订单只到达总线一次，它的事件行随之消失。同一个订单结账两次只发布一个事件：第二次结账失败，它的事件已经回滚。事件行不唤醒任何订阅，所以这些测试用 TestApp::start_live 在真实时钟上运行，越过轮询间隔。