Субъекты, шаблоны и очереди

Это руководство называет субъекты Core NATS, читает многие из них по шаблону, делит работу между репликами в очереди и сохраняет порядок каждого ключа в параллельных партициях воркеров.

Субъект - это имя из токенов, разделённых точками, например orders.eu.created. Core NATS отдаёт сообщение каждой подписке с подходящим субъектом, которая есть в этот момент, и ничего не хранит.

Пример - сервис заказов. Он упаковывает европейские заказы, выставляет по ним счета и записывает в аудит каждое событие заказа во всех регионах.

Очередь

pack читает orders.eu.created. Сервис работает в нескольких репликах, и каждый заказ нужно упаковать один раз.

Макросы

src/main.rs

#[subscriber(CoreSubject::new("orders.eu.created").queue_group("packers"))]
async fn pack(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order packed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Pack;

impl Handle<Order> for Pack {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order packed");
        ready(Ok(()))
    }
}

CoreSubject читает один субъект. queue_group включает подписку в очередь packers. Сервер отдаёт каждое сообщение одному участнику очереди, поэтому каждая реплика упаковывает свою долю заказов. Для сервера одно имя - одна очередь, даже на разных субъектах. Подписка вне очереди или в другой очереди всё равно получает свою копию.

Параллельные партиции воркеров по клиентам

bill читает тот же субъект без очереди, поэтому получает каждый заказ, который получает pack. Он выставляет счета по нескольким заказам одновременно, не больше чем по восьми, и сохраняет порядок заказов каждого клиента.

Макросы

src/main.rs

/// The customer an order belongs to, carried in the partition-key header.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct CustomerKey {
    #[serde(rename = "nats-partition-key")]
    customer: String,
}

/// The invoice of one order, keyed by the customer like the order it bills.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoices.eu", headers = CustomerKey)]
struct Invoice {
    order: u64,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Invoice)]
struct Invoices;

#[subscriber("orders.eu.created", workers(8, by_key))]
async fn bill(
    order: &Order,
    Headers(key): Headers<CustomerKey>,
    Out(invoices): Out<impl Publisher, Invoices>,
) -> HandlerOutcome {
    let invoice = Invoice {
        order: order.id,
        amount_cents: order.amount_cents,
    };
    if invoices
        .message(&invoice)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, customer = %key.customer, "order billed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// The customer an order belongs to, carried in the partition-key header.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct CustomerKey {
    #[serde(rename = "nats-partition-key")]
    customer: String,
}

/// The invoice of one order, keyed by the customer like the order it bills.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Invoice {
    order: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Invoice {
    type Form = FixedName;

    const DESTINATION: &'static str = "invoices.eu";
}

impl MessageHeaders for Invoice {
    type Contract = WithHeaders<CustomerKey>;
}

impl MessageInfo for Invoice {
    const NAME: &'static str = "Invoice";
}

struct Invoices;

impl OutSlot for Invoices {
    const NAME: &'static str = "Invoices";
    type Destination = Reads;
}

impl PublishedThrough<Invoices> for Invoice {}

struct Bill;

impl<Sent> Handle<Message<CustomerKey, Order>, (), Outs<(Sent,)>> for Bill
where
    Sent: OutEntry<Invoices, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Message<CustomerKey, Order>,
        outs: &Outs<(Sent,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let key = &order.headers;
        let invoice = Invoice {
            order: order.body.id,
            amount_cents: order.body.amount_cents,
        };
        if outs
            .get(Invoices)
            .message(&invoice)
            .with_headers(key)
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(order.body.id, customer = %key.customer, "order billed");
        Ok(())
    }
}

workers(8, by_key) запускает восемь партиций воркеров. Доставки с одним ключом выполняются друг за другом в одной партиции воркеров. В ручной форме та же настройка - workers_by_key. Собственного ключа у NATS нет, поэтому ключ передаётся в заголовке nats-partition-key (PARTITION_KEY_HEADER), и его задаёт отправитель. CustomerKey - контракт заголовков с одним полем под этим именем. Headers<CustomerKey> читает ключ доставки, а with_headers(&key) записывает его в счёт. Поэтому счета одного клиента сохраняют порядок и для следующего читателя. Заказ без этого заголовка не проходит проверку контракта, и его доставку завершает политика декодирования, которая по умолчанию отбрасывает сообщение.

Заголовок NATS - это текст: имя состоит из печатных символов ASCII без двоеточия, а значение - одна строка. Публикация с заголовком, который этим правилам не отвечает, возвращает ошибку.

Шаблон

audit читает каждое событие заказа во всех регионах.

Макросы

src/main.rs

#[subscriber(CoreWildcard::new("orders.>"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order event audited");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Audit;

impl Handle<Order> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order event audited");
        ready(Ok(()))
    }
}

CoreWildcard читает каждый субъект, к которому подходит шаблон. * заменяет один токен: orders.*.created читает создание заказов во всех регионах. > заменяет один или несколько последних токенов: orders.> читает каждый субъект внутри orders.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(pack);
            b.include(bill).out(Invoices, Publish).build();
            // A pattern is no subject to publish a retry copy to, so the mount site names one.
            b.include(audit).out_retry(Publish).to("orders.retry");
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(
                subscriber(
                    CoreSubject::new("orders.eu.created").queue_group("packers"),
                    Pack,
                )
                .build(),
            );
            b.include(
                subscriber("orders.eu.created", Bill)
                    .workers_by_key(nonzero!(8usize))
                    .build(),
            )
            .out(Invoices, Publish)
            .build();
            // A pattern is no subject to publish a retry copy to, so the mount site names one.
            b.include(subscriber(CoreWildcard::new("orders.>"), Audit).build())
                .out_retry(Publish)
                .to("orders.retry");
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

Шаблон читает многие субъекты, поэтому у отложенной копии его доставки нет единственного субъекта, куда её отправить. Точка монтирования называет его через out_retry(Publish).to(..), здесь это orders.retry, который шаблон тоже читает. Его называет каждая регистрация на шаблон, даже если её обработчик никогда не просит переобработку, а без него сервис не запускается. Руководство Переобработка и недоставленные сообщения рассказывает о копиях подробнее.

Доставка at-most-once

Core NATS доставляет сообщение тем подпискам, которые существуют в момент его прихода. Сообщение, опубликованное, когда ни одна подписка ему не подходит, пропадает. У доставки нет и подтверждения, поэтому обработчик, который не справился, не получит сообщение от сервера повторно. Если важно каждое сообщение, субъект должен принадлежать стриму: как его читать, показывает руководство Стримы и консьюмеры.

Тесты в файле примера запускают приложение внутри процесса и проверяют копию каждой подписки, шаблон и порядок счетов по каждому клиенту.