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

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

Установка

Cargo.toml

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

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

ruststream - это сам фреймворк. Фича macros даёт атрибуты, json - кодек по умолчанию, а logging выводит журнал сервиса в терминал. ruststream-amqp - брокер AMQP 1.0. Его фича testing добавляет брокер внутри процесса, на котором работают тесты.

У крейта брокера такие фичи, все выключены по умолчанию:

Фича	Что она даёт
rustls	соединения amqps:// по TLS через rustls
native-tls	соединения amqps:// через TLS платформы
transaction	транзакционную публикацию
asyncapi	подробности AMQP 1.0 в генерируемом документе AsyncAPI
testing	брокер внутри процесса, на котором работает TestApp

Под ним работает клиент fe2o3-amqp. Он написан на Rust, поэтому сборке не нужны системные библиотеки.

Локальный брокер

Для разработки достаточно одного ActiveMQ Artemis:

compose.yaml

services:
  artemis:
    image: apache/activemq-artemis:latest-alpine
    environment:
      ARTEMIS_USER: artemis
      ARTEMIS_PASSWORD: artemis
    ports:
      - "5672:5672"

docker compose up -d

Брокер принимает соединения AMQP на localhost:5672, пользователь - artemis, пароль - artemis. Artemis создаёт очередь, когда сервис впервые её читает, поэтому объявлять ничего не нужно.

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

Макросы

src/main.rs

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

#[subscriber(AmqpAddress::queue("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(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |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(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(subscriber(AmqpAddress::queue("orders"), TakeOrder).build());
        },
    )
}

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

take_order читает заказы в JSON из очереди orders. Заказ с суммой он принимает, а заказ без суммы отклоняет. Очередь называет AmqpAddress::queue, а другие виды адресов показывает руководство Очереди и темы. AmqpBroker::new принимает URL брокера и только запоминает его. Соединение открывается при запуске сервиса. Пользователь и пароль из URL служат для входа сервиса в брокер. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

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

docker compose exec artemis /opt/activemq-artemis/bin/artemis producer \
  --protocol amqp --user artemis --password artemis \
  --destination queue://orders --message-count 1 --message '{"id":1,"amount_cents":1250}'

Сервис пишет в журнал order taken с номером заказа. Ctrl+C останавливает его.

Первый тест

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

cargo test

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

Переход с fe2o3-amqp

Сервис на fe2o3-amqp открывает соединение и сеансы, присоединяет связи и пишет цикл приёма. Вот где каждая из этих частей находится в RustStream:

fe2o3-amqp	RustStream
Connection::open	AmqpBroker::new, с sasl и container_id
Session::begin	ничего: у каждой подписки свой сеанс, а издатели делят один общий
Receiver::attach и цикл recv	обработчик #[subscriber(AmqpAddress::..)]
CreditMode::Auto(n)	AmqpAddress::credit
accept, modify, reject	исход обработчика: ack, retry, drop (Подтверждение и гарантии доставки)
auto_accept(true)	Settle::AtMostOnce
Sender::attach и send	ответ или параметр Out, отправленный через политику Publish
динамический получатель для ответов	request (Запросы и ответы)
Controller и Transaction::declare	TransactionalPublish (Публикация «всё или ничего»)

Сервису, который работает с RabbitMQ по AMQP 0.9.1, нужен ruststream-lapin. У двух протоколов общие порт и схема amqp://, но это разные протоколы.