Брокер RabbitMQ¶

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

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

ruststream-lapin запускает сервис RustStream на RabbitMQ поверх lapin, клиента AMQP 0.9.1. Очередь RabbitMQ - разрушающее хранилище: подтверждённая доставка уходит из очереди. Подтверждение, возврат в очередь и отправка в очередь недоставленных сообщений - это кадры протокола, поэтому фреймворк завершает сообщение на самом брокере.

ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-lapin = "0.7"
serde = { version = "1", features = ["derive"] }

LapinBroker::new не выполняет ввод-вывод, поэтому сервис на RabbitMQ собирается тем же макросом #[ruststream::app], что и на любом другом брокере. Рантайм подключает брокер на старте, до открытия подписок. Подключение поглощает брокер и возвращает ConnectedLapinBroker - единственное значение, у которого есть поверхность подписки и публикации.

Тело обработчика называет совместимости: Out<impl Publisher>, Out<impl TransactionalPublisher>, Out<impl RequestReply>. Оно импортирует только прелюдию фреймворка, и поэтому один и тот же обработчик монтируется с любой политикой публикации, которую выберет файл маршрутов. Прелюдию этого крейта импортирует одно тело: то, которое меняет свойство AMQP для одного сообщения. Оно называет тип настроек, а не тип издателя - Out<impl Publisher<Options = LapinPublishOptions>>, - поэтому обработчик по-прежнему не называет издателя.

Файл маршрутов называет значения и импортирует ruststream_lapin::prelude::*, который приносит с собой прелюдию фреймворка. Он добавляет единые для всего семейства имена точки монтирования: Publish, TransactionalPublish и Request - псевдонимы для LapinPublish, ConfirmsPublish и LapinRequest.

crates/ruststream-lapin/examples/lapin_quickstart.rs

use ruststream::nonzero;
use ruststream_lapin::prelude::*;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    println!("got order {}", order.id);
    HandlerOutcome::ack()
}

crates/ruststream-lapin/examples/lapin_quickstart.rs

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

Модель транспорта¶

Подписка читает одну очередь. #[subscriber("orders")] читает очередь orders, а дескриптор RabbitQueue добавляет привязки, типы очередей и предвыборку. Пакетный обработчик называет свой размер в точке монтирования, и крейт собирает пакет на клиенте, потому что в AMQP нет пакета на уровне протокола; смотрите Пакеты.
На стороне публикации имя сообщения - это ключ маршрутизации, а обменник принадлежит политике публикации: пока вы не назовёте другой, это обменник по умолчанию. Смотрите Публикацию.
Завершение доставки нативное: ack отправляет basic.ack, повтор - basic.reject(requeue = true), отбрасывание - basic.reject(requeue = false), который отправляет сообщение в очередь недоставленных, если у очереди задан такой обменник. Отказ - это кадр, который RabbitMQ считает потраченной доставкой, поэтому повтор из обработчика расходует собственный лимит кворумной очереди.
На брокере ничего не объявляется, пока вы не включите это через .declare_topology(true): топологией управляете вы.

Совместимости¶

Какие из необязательных трейтов-совместимостей фреймворка этот брокер реализует нативно:

Совместимость	Нативно	Примечания
`Subscribe`	да	Читает очередь, которую называет подписка; `RabbitQueue` добавляет привязки, тип очереди и предвыборку.
`BatchSubscriber`	да, на клиенте	AMQP присылает по одному `basic.deliver`, поэтому пакет собирается на клиенте до размера, названного в точке монтирования, в пределах окна предвыборки дескриптора и `batch_wait`. Смотрите Пакеты.
`TransactionalPublisher`	да	Оба транзакционных издателя: `.confirms()` буферизует на клиенте и при фиксации дожидается каждого подтверждения, `.server_tx()` использует транзакции канала AMQP. Смотрите Публикацию.
`OwnedTransactions`	да (только confirms)	Транзакция на подтверждениях - это клиентский буфер, поэтому на одном дескрипторе их может быть открыто сколько угодно. `server_tx` переводит в транзакционный режим сам канал, а это состояние канала, существующее ровно в одном экземпляре.
`RequestReply`	да	`LapinRequest` конструирует запросчика поверх direct reply-to с мультиплексированием по `correlation-id`. Смотрите Запрос-ответ.
`Partitioned`	да	Производитель кладёт ключ в заголовок `amqp-partition-key`, и партиции воркеров рантайма читают его; сам AMQP этот заголовок не интерпретирует. Смотрите Метаданные доставки.
`Seekable` + `Positioned`	нет	Очередь не хранит историю, в которую можно было бы вернуться.
`DescribeServer`	да	Сообщает хост подключения и версию AMQP за ним, и именно это записывает документ AsyncAPI. См. Документ AsyncAPI.

Каркас сервиса¶

Сгенерируйте работающую заготовку через cargo generate, по одному шаблону на форму обмена сообщениями:

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

Документация¶

Крейт документирует себя на docs.rs, и этот обзор и есть руководство:

Подписка - два дескриптора очередей, типы очередей, привязки, предвыборка, предел повторов, отложенная повторная доставка, пакеты и метаданные доставки.
Публикация - модель маршрутизации, политики, свойства AMQP для одного сообщения, ответы и транзакции.
Запрос-ответ - RPC поверх direct reply-to.
Документ AsyncAPI - привязки AMQP, которые сервис сообщает о себе.
Тестирование - рабочее приложение под обвязкой TestApp, внутри процесса или против живого брокера.
Эксплуатация - TLS, настройки подключения, объявление по запросу и известные пробелы.

Установка, учебник и остальные брокеры - на сайте RustStream: https://powersemmi.github.io/ruststream/.