Стримы и консьюмеры

Это руководство читает стрим JetStream через durable- и эфемерные консьюмеры, выбирает, с какого места начинает новый консьюмер, читает метаданные доставки и берёт сообщения пакетами.

Стрим хранит сообщения, опубликованные в его субъекты, каждое под своим порядковым номером. Консьюмер читает стрим в своём темпе и подтверждает сообщения по одному.

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

Стрим

Стрим - это состояние сервера. Сервис находит его при запуске и никогда не создаёт, поэтому создайте его один раз утилитой командной строки NATS:

nats stream add ORDERS --subjects 'orders.*' --defaults

ORDERS хранит каждое сообщение, опубликованное в orders.created, orders.paid и любой другой субъект, к которому подходит orders.*. Обычная публикация в такой субъект сохраняется, кто бы её ни отправил. Если подписка называет стрим, которого на сервере нет, сервис не запускается.

Durable-консьюмер

Макросы

src/main.rs

/// What the warehouse is asked to ship. The sequence is the order's place in the stream.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments.requested")]
struct Shipment {
    order: u64,
    sequence: Option<u64>,
}

#[subscriber(
    JetStreamSubject::new("orders.created", "ORDERS").durable("fulfilment"),
    reply
)]
async fn fulfil(order: &Order, Ctx(sequence): Ctx<StreamSequence>) -> Shipment {
    tracing::info!(order.id, ?sequence, "order sent to the warehouse");
    Shipment {
        order: order.id,
        sequence,
    }
}

Вручную

src/main.rs

/// What the warehouse is asked to ship. The sequence is the order's place in the stream.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Shipment {
    order: u64,
    sequence: Option<u64>,
}

impl OutgoingDestination for Shipment {
    type Form = FixedName;

    const DESTINATION: &'static str = "shipments.requested";
}

impl MessageHeaders for Shipment {
    type Contract = NoHeaders;
}

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

struct Fulfil;

impl Handle<Order, Shipment, (), JetStreamContext> for Fulfil {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, JetStreamContext>,
    ) -> impl Future<Output = Result<Shipment, HandlerOutcome>> {
        let sequence = ctx.context(STREAM_SEQUENCE);
        tracing::info!(order.id, ?sequence, "order sent to the warehouse");
        ready(Ok(Shipment {
            order: order.id,
            sequence,
        }))
    }
}

JetStreamSubject принимает субъект, который читает консьюмер, и стрим, который его хранит. Субъект служит фильтром консьюмера, поэтому fulfil читает из ORDERS только orders.created. Если фильтр должен отличаться от субъекта, его задаёт filter_subject. Сервис создаёт консьюмер при запуске со всеми настройками дескриптора.

durable даёт консьюмеру имя fulfilment. Сервер хранит его позицию, поэтому перезапущенный сервис продолжает с того места, где остановился. Реплики сервиса делят один durable-консьюмер, и каждое сообщение достаётся одной из них. Это аналог очереди в JetStream.

Обработчик читает место доставки в стриме через Ctx<StreamSequence>. Контекст JetStream содержит ещё имена стрима и консьюмера, порядковый номер в консьюмере, счётчик доставок и число сообщений, которые ещё ждут выдачи. В ручной форме обработчик принимает Context<'_, JetStreamContext> и читает те же ключи через ctx.context(..). На доставке Core NATS каждый ключ возвращает None.

Эфемерный консьюмер

Макросы

src/main.rs

#[subscriber(JetStreamSubject::new("orders.*", "ORDERS").deliver_policy(DeliverPolicy::New))]
async fn show(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order event on the dashboard");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Show;

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

Дескриптор без durable создаёт эфемерный консьюмер. У каждого процесса он свой, и сервер удаляет его после закрытия подписки. Поэтому каждая реплика сервиса показывает каждое событие заказа.

DeliverPolicy задаёт, с какого места начинает новый консьюмер. All, значение по умолчанию, начинает с первого сообщения в стриме, а New - со следующего опубликованного. Last, LastPerSubject, ByStartSequence и ByStartTime начинают с последнего сообщения, с последнего сообщения каждого субъекта, с порядкового номера или с момента времени. Политика действует один раз, при создании консьюмера. Дальше durable-консьюмер продолжает со своей позиции.

Пакеты

Макросы

src/main.rs

#[subscriber(
    JetStreamSubject::new("orders.paid", "ORDERS")
        .durable("reconciler")
        .pull_expires(NonZeroDuration::from_secs(nonzero!(1)))
)]
async fn reconcile(orders: &[Order]) -> HandlerOutcome {
    let total: u64 = orders.iter().map(|order| order.amount_cents).sum();
    tracing::info!(orders = orders.len(), total, "paid orders reconciled");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Reconcile;

impl Handle<[Order]> for Reconcile {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let total: u64 = orders.iter().map(|order| order.amount_cents).sum();
        tracing::info!(orders = orders.len(), total, "paid orders reconciled");
        ready(Ok(()))
    }
}

Обработчик, который принимает &[Order], получает пакет сообщений и выполняется один раз на весь пакет. Возвращённый исход завершает доставку каждого сообщения пакета.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(fulfil);
            b.include(show);
            b.include(reconcile.batch(nonzero!(100)));
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.created", "ORDERS").durable("fulfilment"),
                    Fulfil,
                )
                .reply()
                .build(),
            );
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.*", "ORDERS").deliver_policy(DeliverPolicy::New),
                    Show,
                )
                .build(),
            );
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.paid", "ORDERS")
                        .durable("reconciler")
                        .pull_expires(NonZeroDuration::from_secs(nonzero!(1))),
                    Reconcile,
                )
                .batch(nonzero!(100usize))
                .build(),
            );
        },
    )
}

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

Размер пакета задаёт точка монтирования: batch(100) просит у сервера до ста сообщений за одну выборку. pull_expires в дескрипторе задаёт, сколько выборка их ждёт, здесь одну секунду. После этого обработчик получает пакет меньшего размера. По умолчанию выборка ждёт пять секунд.

Ответ fulfil публикуется через Core NATS, как у брокера по умолчанию. Руководство Публикация в стримы сохраняет ответы в стриме. Подписки Core и JetStream работают через одно соединение, поэтому один сервис свободно сочетает обе модели.

max_ack_pending ограничивает, сколько доставок консьюмер держит без подтверждения. По умолчанию это 1024. Как завершается каждая доставка, рассказывает руководство Подтверждение и гарантии доставки.