Начало работы

Это руководство устанавливает ruststream-lapin, запускает локальный RabbitMQ и первый сервис на нём.

Установка

Cargo.toml

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

[dev-dependencies]
ruststream-lapin = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt"] }

ruststream - это сам фреймворк. Фича macros даёт атрибуты, json - кодек по умолчанию, asyncapi - документ AsyncAPI сервиса, а logging выводит журнал сервиса в терминал. ruststream-lapin - брокер RabbitMQ на lapin, клиенте AMQP 0.9.1, написанном на Rust. Для сборки не нужно ничего, кроме Cargo. Фича testing брокера добавляет RabbitMQ внутри процесса, на котором работают тесты.

У крейта брокера такие фичи:

Фича	Что она даёт
asyncapi	привязки AMQP в генерируемом документе AsyncAPI
testing	RabbitMQ внутри процесса, на котором работает TestApp
tls-native-tls	TLS через TLS-библиотеку платформы
tls-rustls	TLS через rustls
tls-rustls-ring	TLS через rustls с криптопровайдером ring
plugin-consistent-hash	обменник с консистентным хешированием из плагина RabbitMQ
plugin-dme	отложенная переобработка через плагин delayed-message-exchange

Локальный RabbitMQ

Для разработки достаточно одного узла RabbitMQ с веб-интерфейсом управления:

compose.yaml

services:
  rabbitmq:
    image: rabbitmq:4-management-alpine
    ports:
      - "127.0.0.1:5672:5672"
      - "127.0.0.1:15672:15672"

docker compose up -d

Брокер слушает localhost:5672. Веб-интерфейс управления открывается по адресу http://localhost:15672, пользователь guest, пароль guest. RabbitMQ пускает учётную запись guest только с локального адреса, поэтому порты опубликованы на 127.0.0.1.

Первый сервис

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[subscriber("orders")]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672").declare_topology(true),
        |b| {
            b.include(take_order);
        },
    )
}

Вручную

src/main.rs

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

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672").declare_topology(true),
        |b| {
            b.include(subscriber("orders", TakeOrder).build());
        },
    )
}

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

take_order читает заказы в JSON из очереди orders. Заказ с суммой он подтверждает, а заказ без суммы отбрасывает. LapinBroker::new принимает URI соединения и только запоминает его. Соединение открывается при запуске сервиса. declare_topology(true) разрешает сервису объявить очереди, которые он читает, поэтому сервис запускается и на пустом RabbitMQ. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Опубликуйте заказ из другого терминала утилитой rabbitmqadmin, которая входит в образ:

docker compose exec rabbitmq rabbitmqadmin publish message \
  --routing-key orders --payload '{"id":1,"amount_cents":1250}'

Сообщение уходит в обменник по умолчанию, а он доставляет его в очередь, имя которой совпадает с ключом маршрутизации. Сервис пишет в журнал order taken с номером заказа. То же сообщение можно опубликовать со страницы очереди orders в веб-интерфейсе управления. Ctrl+C останавливает сервис.

Сервис умеет и описать себя:

cargo run -- asyncapi gen

Команда выводит документ AsyncAPI сервиса: сервер, канал orders и сообщение Order.

cargo generate создаёт такой же сервис с роутером и ответом:

cargo generate --git https://github.com/powersemmi/ruststream-lapin templates/amqp-queue --name my-service

Шаблон templates/amqp-topic вместо этого начинает с обменника типа topic.

Первый тест

В файле примера есть и два теста. Каждый запускает то же app через TestApp::start, публикует заказ и проверяет, как take_order подтвердил его доставку. Там LapinBroker подключается к RabbitMQ внутри процесса, поэтому тестам не нужен сервер:

cargo test

Руководство Тестирование сервиса на RabbitMQ разбирает эти тесты и то, что моделирует RabbitMQ внутри процесса.

Переход с lapin

Сервис на lapin сам открывает каналы и пишет цикл потребителя. Вот куда переходит каждая часть в RustStream:

lapin	RustStream
Connection::connect и create_channel	LapinBroker::new; рантайм подключается при запуске
channel.basic_qos	prefetch у брокера или у отдельной очереди
queue_declare, exchange_declare, queue_bind	дескриптор очереди и declare_topology(true) (Топология)
basic_consume и цикл while let Some(delivery)	обработчик #[subscriber]
десериализация delivery.data	тип параметра обработчика, который декодирует кодек
delivery.ack и delivery.reject	исход обработчика (Подтверждения, переобработка и недоставленные сообщения)
basic_publish	ответ или параметр Out, отправленный через политику публикации
confirm_select и ожидание подтверждений	ConfirmsPublish (Надёжная публикация)
очередь для ответов и таблица корреляции	LapinRequest (Запрос и ответ)