Учебник: собираем первый сервис

Этот учебник собирает сервис заказов с нуля и разбирает каждую его часть. Сервис работает на in-memory брокере из ядра, поэтому ему не нужны ни сервер, ни адаптация библиотеки брокера. Переход на настоящий брокер - правка в одну строку, её показывает шаг 7.

1. Создайте крейт

cargo new orders-service
cd orders-service

Cargo.toml

[package]
name = "orders-service"
version = "0.1.0"
edition = "2024"

[dependencies]
ruststream = { version = "0.7", features = ["macros", "memory", "json", "asyncapi", "logging"] }
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

2. Опишите сообщение и обработчик

Обработчик - это async fn. Первым параметром он получает полезную нагрузку, уже декодированную из сообщения. Макрос #[subscriber] делает из функции определение подписчика с тем же именем.

Макросы

src/orders.rs

use ruststream::runtime::HandlerOutcome;
use ruststream::schemars::JsonSchema;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

/// An order placed by a customer.
#[derive(Debug, Deserialize, JsonSchema)]
pub(crate) struct Order {
    pub(crate) id: u64,
    pub(crate) quantity: u32,
}

#[subscriber("orders")]
pub(crate) async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.quantity, "order received");
    HandlerOutcome::ack()
}

Вручную

src/orders.rs

use std::future::{Future, ready};

use ruststream::prelude::*;
use ruststream::schemars::JsonSchema;
use serde::Deserialize;

/// An order placed by a customer.
#[derive(Debug, Deserialize, JsonSchema)]
pub(crate) struct Order {
    pub(crate) id: u64,
    pub(crate) quantity: u32,
}

/// The handler: `#[subscriber("orders")]` generates this struct and this impl. The subject it
/// carried is named where the handler is mounted instead.
pub(crate) struct Receive;

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

Обработчик возвращает исход, HandlerOutcome:

• ack подтверждает сообщение;
• nack отбрасывает его или возвращает в очередь.

Вместо исхода можно вернуть () или Result<(), E>. Тогда Ok подтверждает сообщение, а Err отбрасывает.

JsonSchema добавляет схему полезной нагрузки в AsyncAPI-документ из шага 6, а doc-комментарий типа становится описанием сообщения. Фича asyncapi реэкспортирует schemars, поэтому отдельная зависимость не нужна.

3. Свяжите обработчик с приложением

Приложение - это функция с атрибутом #[ruststream::app]. В ней with_broker подключает брокер, а include регистрирует на нём обработчик:

Макросы

src/main.rs

mod orders;

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

use crate::orders::handle;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
    })
}

Вручную

src/main.rs

mod orders;

use std::process::ExitCode;

use ruststream::memory::prelude::*;
use ruststream::runtime::cli::run_main;

use crate::orders::Receive;

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Receive).build());
    })
}

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

Запустите сервис:

cargo run -- run

Фича logging выводит журнал сервиса в терминал: сначала строки запуска, затем по строке на каждый заказ, который обработчик записывает через tracing::info!.

Кодек по умолчанию

include декодирует сообщения кодеком по умолчанию. Это json, если фича включена, иначе cbor, иначе msgpack. Другой кодек для всех обработчиков брокера задаётся один раз через with_broker_codec(broker, codec, |b| ...). Правила выбора кодека описаны в разделе Кодеки.

4. Ответьте на сообщения

Переключить обработчик в reply-форму можно, указав аргумент reply. Значение, которое возвращает обработчик в reply-форме, публикуется как ответ. Тип ответа выводит Outgoing. Тип без этого derive не компилируется.

Адрес ответа задаётся одним из двух способов:

• В самом аргументе: #[subscriber("orders", reply("confirmations"))]. Тогда тип ответа выводит Outgoing без имени.
• На типе ответа: #[outgoing(name = "confirmations")]. Тогда хватает аргумента reply без значения.

Если адрес объявлен на типе, он важнее адреса в reply("..."). Пример ниже объявляет адрес на типе:

Макросы

src/orders.rs

/// The service's answer to an order.
#[derive(Debug, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "confirmations")]
pub(crate) struct Confirmation {
    pub(crate) id: u64,
    pub(crate) accepted: bool,
}

#[subscriber("orders", reply)]
pub(crate) async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

Вручную

src/orders.rs

use serde::Serialize;

/// The service's answer to an order, published on `confirmations`.
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct Confirmation {
    pub(crate) id: u64,
    pub(crate) accepted: bool,
}

/// What `#[derive(Outgoing)]` with `#[outgoing(name = "confirmations")]` writes: the destination
/// is fixed on the type, so nothing else names it.
impl OutgoingDestination for Confirmation {
    type Form = FixedName;

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

impl MessageHeaders for Confirmation {
    type Contract = NoHeaders;
}

/// The reply form of the same trait: the second parameter of `Handle` is the reply type, so the
/// body returns a `Confirmation`, and the mount site supplies the publisher it leaves through.
pub(crate) struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation {
            id: order.id,
            accepted: order.quantity > 0,
        }))
    }
}

Подключите confirm тем же include, что и handle:

Макросы

src/main.rs

use crate::orders::{confirm, handle};

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
        b.include(confirm);
    })
}

Вручную

src/main.rs

use crate::orders::{Confirm, Receive};

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Receive).build());
        // The definition names the reply's destination; with no `.out_reply(..)` at the mount
        // site the reply leaves through the broker's default publisher.
        b.include(
            subscriber("orders", Confirm)
                .reply()
                .to("confirmations")
                .build(),
        );
    })
}

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

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

5. Наведите порядок роутером

Когда обработчиков становится много, вынесите их в отдельный модуль и соберите в Router. Роутер - это набор обработчиков, который приложение подключает к брокеру одним вызовом.

Макросы

src/routes.rs

use ruststream::memory::prelude::*;

use crate::orders;

// The reply wiring is a publish policy: pure declaration, so the router needs no broker at all.
pub(crate) fn orders() -> impl RouterDef<MemoryBroker> {
    Router::new()
        .include(orders::handle)
        .include(orders::confirm)
        .out_reply(Publish)
        .build()
}

Вручную

src/routes.rs

use ruststream::memory::prelude::*;

use crate::orders::{Confirm, Receive};

// Each handler is bound to its subject where it is mounted; the definition says what it replies
// with, the reply type says where it goes, and the mount chain names who publishes it. The
// publisher wiring is still a publish policy - pure declaration, so the router needs no broker
// at all.
pub(crate) fn orders() -> impl RouterDef<MemoryBroker> {
    Router::new()
        .include(subscriber("orders", Receive).build())
        .include(subscriber("orders", Confirm).reply().build())
        .out_reply(Publish)
        .build()
}

include добавляет в роутер обычный обработчик. Для обработчика в reply-форме он возвращает цепочку:

• .out_reply(..) задаёт политику публикации ответа;
• .build() завершает регистрацию.

Без .out_reply(..) метод .build() берёт политику брокера по умолчанию, ту же, что include в шаге 4.

Приложение подключает роутер через include_router:

Макросы

src/main.rs

mod orders;
mod routes;

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        let router = routes::orders();
        b.include_router(router);
    })
}

Вручную

src/main.rs

mod orders;
mod routes;

use std::process::ExitCode;

use ruststream::memory::prelude::*;
use ruststream::runtime::cli::run_main;

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        let router = routes::orders();
        b.include_router(router);
    })
}

// What `#[ruststream::app]` writes around the builder: the `run` and `asyncapi gen` commands.
fn main() -> ExitCode {
    run_main(app)
}

Остальные возможности роутера описаны в разделе Роутинг.

6. Посмотрите AsyncAPI-документ

cargo run -- asyncapi gen

Команда печатает AsyncAPI-документ сервиса. Каждый подписчик добавляет в него канал и операцию receive. handle и confirm слушают канал orders, но у каждого своя подписка и своя операция. Ответ добавляет операцию send на канале confirmations.

Схемы полезных нагрузок лежат в components.messages. Флаги вывода (-o, --yaml) и устройство документа описаны в руководстве по AsyncAPI.

7. Перейдите на настоящий брокер

Ничто из написанного выше не привязано к in-memory брокеру. Чтобы перейти на другой брокер, добавьте в зависимости адаптацию его клиентской библиотеки и создайте брокер в with_broker вместо MemoryBroker::new(), например NatsBroker::new("nats://localhost:4222"). Это одна строка. Обработчики, роутер и кодеки остаются прежними. Список адаптаций и замену для каждой из них даёт раздел Брокеры.

Готовый сервис - это компилируемый пример

Каждый фрагмент этой страницы взят из examples/tutorial, который CI собирает при каждом изменении. first_app.rs и reply_app.rs - это сервис после шагов 3 и 4, а main.rs - готовый. Запустить его можно командой cargo run --example tutorial --features macros,memory,json,asyncapi,logging -- run.

Что дальше

• Middleware - сквозная логика вокруг обработчиков.
• Жизненный цикл - разделяемое состояние и хуки старта и остановки.
• Тестирование - тесты только что написанных обработчиков прямо в процессе.
• Метрики - счётчики и гистограммы Prometheus.