Публикация в стримы

Это руководство публикует ответы и сообщения в стримы JetStream, сохраняет повторное сообщение один раз и отклоняет запись, сделанную по устаревшей версии.

Обычная публикация уходит через Core NATS, и никто её не подтверждает. JetStreamPublish ждёт, пока стрим подтвердит каждое сообщение. Поэтому сообщение, которое стрим отклонил, - это ошибка, которую видит обработчик, а не тихая потеря.

Пример - сервис заказов. Он подтверждает каждый заказ, архивирует его и хранит историю каждой учётной записи клиента. Стримы, в которые он пишет, уже есть на сервере:

nats stream add ORDERS --subjects 'orders.placed' --defaults
nats stream add CONFIRMATIONS --subjects 'confirmations.orders' --defaults
nats stream add ARCHIVE --subjects 'archive.orders' --defaults
nats stream add ACCOUNTS --subjects 'accounts.*' --defaults

Ответ в стрим

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "confirmations.orders")]
struct Confirmation {
    order: u64,
}

#[subscriber(
    JetStreamSubject::new("orders.placed", "ORDERS").durable("confirmer"),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { order: order.id }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Confirmation {
    order: u64,
}

impl OutgoingDestination for Confirmation {
    type Form = FixedName;

    const DESTINATION: &'static str = "confirmations.orders";
}

impl MessageHeaders for Confirmation {
    type Contract = NoHeaders;
}

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

struct Confirm;

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

confirm отвечает на каждый заказ сообщением Confirmation в confirmations.orders. Приложение ниже отправляет этот ответ через JetStreamPublish::default().expect_stream("CONFIRMATIONS"). expect_stream называет стрим, который должен хранить субъект, и публикация, которую принял бы другой стрим, отклоняется. Заказ подтверждается только после того, как стрим сохранил подтверждение. Если подтверждение отклонено, доставка заказа завершается ошибкой, и заказ возвращается.

Субъект ответа лежит вне субъектов ORDERS. Ответ, сохранённый в стриме, который читает обработчик, снова пришёл бы к этому обработчику.

Однократное сохранение

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "archive.orders")]
struct Archived {
    order: u64,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Archived)]
struct Archive;

#[subscriber(JetStreamSubject::new("orders.placed", "ORDERS").durable("archiver"))]
async fn archive(
    order: &Order,
    Out(out): Out<impl Publisher<Options = JetStreamOptions>, Archive>,
) -> HandlerOutcome {
    let archived = Archived {
        order: order.id,
        amount_cents: order.amount_cents,
    };
    // The same id on every delivery of this order: the stream stores the first copy only.
    if out
        .message(&archived)
        .message_id(format!("order-{}", order.id))
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[subscriber(JetStreamSubject::new("archive.orders", "ARCHIVE").durable("indexer"))]
async fn index(archived: &Archived) -> HandlerOutcome {
    tracing::info!(archived.order, "archived order indexed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Archived {
    order: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Archived {
    type Form = FixedName;

    const DESTINATION: &'static str = "archive.orders";
}

impl MessageHeaders for Archived {
    type Contract = NoHeaders;
}

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

struct Archive;

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

impl PublishedThrough<Archive> for Archived {}

struct ArchiveOrder;

impl<Copies> Handle<Order, (), Outs<(Copies,)>> for ArchiveOrder
where
    Copies: OutEntry<Archive, Wire: Publisher<Options = JetStreamOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Copies,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let archived = Archived {
            order: order.id,
            amount_cents: order.amount_cents,
        };
        // The same id on every delivery of this order: the stream stores the first copy only.
        if outs
            .get(Archive)
            .message(&archived)
            .message_id(format!("order-{}", order.id))
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

struct Index;

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

Заказ может быть доставлен дважды, и тогда archive публикует его копию дважды. message_id даёт каждой копии одного заказа один и тот же идентификатор, который передаётся в заголовке Nats-Msg-Id. Стрим сохраняет первую копию и подтверждает повтор, не сохраняя его, поэтому index читает каждый заказ один раз. Стрим помнит идентификаторы в течение своего окна дубликатов: две минуты, если стрим не задаёт другое значение.

message_id описывает одно сообщение, поэтому это шаг публикации, а не настройка точки монтирования. Шаги даёт JetStreamPublishSteps из прелюдии крейта. Тело обработчика, которое вызывает такой шаг, ограничивает свой слот типом Out<impl Publisher<Options = JetStreamOptions>, _>. Это ограничение привязывает обработчик к JetStream. Только здесь обработчик называет брокер.

Отказ в устаревшей записи

Макросы

src/main.rs

/// A change of an account's e-mail, made from the version of the account the client last read:
/// the stream sequence of its last event, or 0 for an account with no events yet.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "commands.change-email")]
struct ChangeEmail {
    account: u64,
    version: u64,
    email: String,
}

/// An event of an account's history, stored under the account's own subject.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct EmailChanged {
    account: u64,
    email: String,
}

#[derive(OutSlot)]
#[publishes(EmailChanged)]
struct History;

#[subscriber("commands.change-email")]
async fn change_email(
    change: &ChangeEmail,
    Out(history): Out<impl Publisher<Options = JetStreamOptions>, History>,
) -> HandlerOutcome {
    let event = EmailChanged {
        account: change.account,
        email: change.email.clone(),
    };
    let stored = history
        .message(&event)
        .to(format!("accounts.{}", change.account))
        .expect_last_subject_sequence(change.version)
        .publish()
        .await;
    if stored.is_err() {
        tracing::warn!(
            change.account,
            change.version,
            "change made from a stale version refused"
        );
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// A change of an account's e-mail, made from the version of the account the client last read:
/// the stream sequence of its last event, or 0 for an account with no events yet.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ChangeEmail {
    account: u64,
    version: u64,
    email: String,
}

impl OutgoingDestination for ChangeEmail {
    type Form = FixedName;

    const DESTINATION: &'static str = "commands.change-email";
}

impl MessageHeaders for ChangeEmail {
    type Contract = NoHeaders;
}

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

/// An event of an account's history, stored under the account's own subject.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct EmailChanged {
    account: u64,
    email: String,
}

impl OutgoingDestination for EmailChanged {
    type Form = CallerName;
}

impl MessageHeaders for EmailChanged {
    type Contract = NoHeaders;
}

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

struct History;

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

impl PublishedThrough<History> for EmailChanged {}

struct ApplyChange;

impl<Events> Handle<ChangeEmail, (), Outs<(Events,)>> for ApplyChange
where
    Events: OutEntry<History, Wire: Publisher<Options = JetStreamOptions>>,
{
    async fn handle(
        &self,
        change: &ChangeEmail,
        outs: &Outs<(Events,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let event = EmailChanged {
            account: change.account,
            email: change.email.clone(),
        };
        let stored = outs
            .get(History)
            .message(&event)
            .to(format!("accounts.{}", change.account))
            .expect_last_subject_sequence(change.version)
            .publish()
            .await;
        if stored.is_err() {
            tracing::warn!(
                change.account,
                change.version,
                "change made from a stale version refused"
            );
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

История учётной записи 42 - это субъект стрима accounts.42. Клиент меняет адрес электронной почты учётной записи, опираясь на версию, которую прочитал последней: порядковый номер последнего события учётной записи в стриме. expect_last_subject_sequence заставляет стрим сохранить событие, только если последнее сообщение субъекта всё ещё стоит под этим номером. Два клиента, прочитавшие одну версию, не могут записать оба: вторая запись отклоняется, и change_email её отбрасывает. expect_last_sequence проверяет весь стрим, а не один субъект, а expect_last_message_id - идентификатор последнего сообщения стрима.

Приложение

Макросы

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(confirm)
                .out_reply(JetStreamPublish::default().expect_stream("CONFIRMATIONS"));
            b.include(archive)
                .out(
                    Archive,
                    JetStreamPublish::default().expect_stream("ARCHIVE"),
                )
                .build();
            b.include(index);
            b.include(change_email)
                .out(
                    History,
                    JetStreamPublish::default().expect_stream("ACCOUNTS"),
                )
                .build();
        },
    )
}

Вручную

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(
                    JetStreamSubject::new("orders.placed", "ORDERS").durable("confirmer"),
                    Confirm,
                )
                .reply()
                .build(),
            )
            .out_reply(JetStreamPublish::default().expect_stream("CONFIRMATIONS"));
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.placed", "ORDERS").durable("archiver"),
                    ArchiveOrder,
                )
                .build(),
            )
            .out(
                Archive,
                JetStreamPublish::default().expect_stream("ARCHIVE"),
            )
            .build();
            b.include(
                subscriber(
                    JetStreamSubject::new("archive.orders", "ARCHIVE").durable("indexer"),
                    Index,
                )
                .build(),
            );
            b.include(subscriber("commands.change-email", ApplyChange).build())
                .out(
                    History,
                    JetStreamPublish::default().expect_stream("ACCOUNTS"),
                )
                .build();
        },
    )
}

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

Каждая публикация JetStream ждёт подтверждения от стрима. Вне доставки хук запуска принимает политику, и publish_ack возвращает это подтверждение: стрим, порядковый номер и то, видело ли сообщение окно дубликатов. Публикация Core NATS подходит там, где потеря сообщения допустима, например для панели мониторинга из руководства Подтверждение и гарантии доставки.

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