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

Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Этот учебник собирает сервис заказов с нуля и разбирает каждую его часть. Сервис работает на 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"] }
serde = { version = "1", features = ["derive"] }

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 {
    println!("order {} x{}", order.id, order.quantity);
    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>> {
        println!("order {} x{}", order.id, order.quantity);
        ready(Ok(()))
    }
}

Обработчик возвращает HandlerOutcome: либо ack, либо nack, который отбрасывает сообщение или возвращает его в очередь. Вместо исхода можно вернуть () или Result<(), E>, где Ok подтверждает, а Err отбрасывает.

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

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

МакросыВручную

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::error::Error;

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

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());
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

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

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

Запустите:

cargo run -- run

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

Чтобы опубликовать ответ, верните его из обработчика и напишите у подписчика publish. Адресата объявляет derive Outgoing на типе ответа:

МакросыВручную

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", publish)]
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 рядом с handle тем же include. Ответ публикуется политикой публикации брокера по умолчанию и кодируется кодеком по умолчанию.

МакросыВручную

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(),
        );
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

Публикацию изнутри обработчика и остальные способы разбирает раздел Публикация и ответы.

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()
}

Обработчик с ответом монтируется на роутер цепочкой: .out_reply(..) задаёт политику публикации ответа, а .build() фиксирует регистрацию. Без .out_reply(..) .build() берёт ту же политику публикации брокера по умолчанию, что и include в шаге 4. Остальные возможности роутера разбирает раздел Роутинг.

МакросыВручную

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::error::Error;

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

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]` wraps around the builder: the runtime entry point. Its CLI
// (`run`, `asyncapi gen`) is what a hand-written `main` gives up.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

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

cargo run -- asyncapi gen

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

Схемы полезных нагрузок документ хранит в 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 -- run.

Что дальше¶

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