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

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

Установка

Cargo.toml

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

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

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

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

Фича	Что она даёт
testing	NATS внутри процесса, на котором работает TestApp
asyncapi	привязки NATS в генерируемом документе AsyncAPI

Крейт построен на клиенте async-nats. Он написан на Rust и шифрует соединение через rustls, поэтому сборке не нужны системные библиотеки.

Локальный NATS

Для разработки достаточно одного сервера NATS с JetStream:

compose.yaml

services:
  nats:
    image: nats:2-alpine
    command: ["-js"]
    ports:
      - "4222:4222"

docker compose up -d

Сервер слушает localhost:4222. Флаг -js включает JetStream. С его стримами работают руководства, начиная со Стримов и консьюмеров. Примеры публикуют сообщения утилитой командной строки NATS nats.

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

Макросы

src/main.rs

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

#[subscriber("orders.created")]
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(
        NatsBroker::new("nats://localhost:4222"),
        |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(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(subscriber("orders.created", TakeOrder).build());
        },
    )
}

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

take_order читает заказы в JSON из субъекта Core NATS orders.created. Заказ с суммой он принимает, а заказ без суммы отбрасывает. NatsBroker::new принимает адрес сервера и только запоминает его. Соединение открывается при запуске сервиса. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Опубликуйте заказ из другого терминала:

nats pub orders.created '{"id":1,"amount_cents":1250}'

Сервис пишет в журнал order taken с номером заказа. В Core NATS нет подтверждения, поэтому для каждого заказа в журнале есть и ack / nack failed: рантайм сообщает, что завершить доставку не удалось. Руководство Подтверждение и гарантии доставки объясняет, когда это важно и как стрим завершает каждое сообщение. Ctrl+C останавливает сервис.

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

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

Шаблон templates/nats-js создаёт сервис, который читает стрим через durable-консьюмер.

Первый тест

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

cargo test

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

Переход с async-nats

Сервис на async-nats сам пишет цикл для каждой подписки и публикует сообщения вручную. Вот куда переходит каждая часть в RustStream:

async-nats	RustStream
async_nats::connect или ConnectOptions::connect	NatsBroker::new и with_options для ConnectOptions
client.subscribe и цикл while let Some(message)	обработчик #[subscriber("subject")]
client.queue_subscribe	CoreSubject::queue_group (Субъекты, шаблоны и очереди)
serde_json::from_slice(&message.payload)	тип параметра обработчика, который декодирует кодек
client.publish	ответ или параметр Out, отправленный через политику Publish
client.request	request на слоте Out<impl RequestReply, _> (Запрос и ответ)
create_consumer на стриме и цикл messages()	JetStreamSubject (Стримы и консьюмеры)
message.ack(), AckKind::Nak, AckKind::Term	исход обработчика: ack, retry, retry_after, drop (Подтверждение и гарантии доставки)
jetstream.publish и его подтверждение	JetStreamPublish (Публикация в стримы)