Сервис и его первый подписчик

Сначала магазин - это один подписчик, который принимает каждый оформленный заказ. Небольшое приложение монтирует его на in-memory брокере.

Сервис на RustStream - это приложение: билдер, который перечисляет брокеры сервиса и монтирует на них подписчиков. Подписчик - это async fn, первый параметр которой - сообщение. Макрос #[subscriber] задаёт тему, которую он читает. Рантайм декодирует полезную нагрузку до запуска тела обработчика, по умолчанию из JSON. Значение, которое возвращает тело, завершает доставку: HandlerOutcome::ack сообщает брокеру, что сообщение обработано. In-memory брокер передаёт сообщения внутри процесса, поэтому магазину ничего не нужно устанавливать. Глава 11 заменяет его сетевым брокером, а подписчики остаются прежними.

Заказ

Заказ - это тип serde. Derive-макрос Outgoing задаёт тему, в которую публикуется заказ: orders.placed. Сам магазин пока не публикует заказы. Это делает тест, а в главе 10 - HTTP-витрина.

Макросы

examples/guide/first.rs

/// An order placed in the storefront.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders.placed")]
struct Order {
    id: u64,
    customer: String,
}

Вручную

examples/manual/guide/first.rs

/// An order placed in the storefront.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Order {
    id: u64,
    customer: String,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "orders.placed")]` writes.
impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders.placed";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

Ручная форма содержит две реализации, которые генерирует derive-макрос: адрес публикации и заголовки заказа (пока их нет).

Первый подписчик

accept читает тему orders.placed и записывает каждый заказ в журнал через tracing. Ручная форма - это именованный тип с реализацией impl Handle<Order>: Ok(()) подтверждает сообщение, а Err(outcome) завершает доставку этим исходом.

Макросы

examples/guide/first.rs

/// Takes every placed order; for now it only logs it.
#[subscriber("orders.placed")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order = order.id, customer = %order.customer, "order accepted");
    HandlerOutcome::ack()
}

Вручную

examples/manual/guide/first.rs

/// Takes every placed order; for now it only logs it.
struct Accept;

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

Приложение

RustStream - это приложение. with_broker регистрирует брокер и монтирует на нём подписчиков через include. Билдер только запоминает переданное: брокер подключается при запуске сервиса. Атрибут #[ruststream::app] превращает билдер в main с командами run и asyncapi gen. В ручной форме этот main написан явно: run_main поверх билдера.

Макросы

examples/guide/first.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
    })
}

Вручную

examples/manual/guide/first.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders.placed", Accept).build());
    })
}

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

Запуск

cargo run --example guide_first --features testing,macros,memory,json,logging -- run

Фича logging выводит журнал сервиса в терминал. Сервис подключает брокер, создаёт подписку на тему и работает до нажатия Ctrl+C:

INFO ruststream::lifecycle starting service service=shop version=0.1.0 brokers=1 subscribers=1
INFO ruststream::lifecycle broker connected broker="ruststream::memory::MemoryBroker"
INFO ruststream::dispatch subscriber started subscriber=orders.placed input="guide_first::Order"
INFO ruststream::lifecycle service running subscribers=1

Остальные главы превращают этот файл в магазин из examples/guide.