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

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

Установка

Cargo.toml

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

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

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

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

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

Системные требования

Клиент Pulsar компилирует описания протокола Pulsar во время сборки, поэтому для сборки нужен protoc, компилятор Protocol Buffers. В Debian и Ubuntu его ставит пакет protobuf-compiler, а в macOS - команда brew install protobuf. Путь к protoc, который не лежит в PATH, задаёт переменная окружения PROTOC.

Локальный Pulsar

Pulsar в автономном режиме (standalone) запускает брокер и его хранилище в одном контейнере:

compose.yaml

services:
  pulsar:
    image: apachepulsar/pulsar:4.1.2
    command: bin/pulsar standalone
    ports:
      - "6650:6650"
      - "8080:8080"
    environment:
      # The address the broker hands to clients outside the container.
      PULSAR_PREFIX_advertisedAddress: 127.0.0.1
      PULSAR_PREFIX_advertisedListeners: pulsar://127.0.0.1:6650
    healthcheck:
      test: ["CMD-SHELL", "curl -fsS http://localhost:8080/admin/v2/brokers/health || exit 1"]
      interval: 10s
      retries: 30

Запустите его и дождитесь, пока он станет работоспособным:

docker compose up -d --wait

Брокер слушает localhost:6650, а API администрирования - localhost:8080. Pulsar создаёт топик при первом обращении к нему, поэтому сервису не нужна предварительная настройка.

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

Макросы

src/main.rs

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

#[subscriber(PulsarSubscription::new("orders", "billing"))]
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("billing", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(take_order);
        },
    )
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
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("billing", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(subscriber(PulsarSubscription::new("orders", "billing"), TakeOrder).build());
        },
    )
}

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

take_order читает заказы в JSON из топика orders. Заказ с суммой он подтверждает, а заказ без суммы отбрасывает. Pulsar читает топик только через именованную подписку. PulsarSubscription::new называет топик и подписку, здесь billing. Подписка - это курсор, который хранит сервер, поэтому после перезапуска сервис продолжает с того места, где остановился. PulsarBroker::new принимает адрес подключения (service URL) и только запоминает его. Соединение открывается при запуске сервиса. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

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

docker compose exec pulsar bin/pulsar-client produce orders \
  -s ';' -m '{"id":1,"amount_cents":1250}'

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

Первый тест

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

cargo test

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

Переход с крейта pulsar

Сервис на крейте pulsar сам строит клиент, цикл консьюмера и продюсеры. Вот куда переходит каждая часть в RustStream:

Крейт pulsar	RustStream
Pulsar::builder(url, TokioExecutor).build().await	PulsarBroker::new(url), который подключает рантайм
.with_auth(Authentication { name: "token", .. })	PulsarBroker::token
.with_operation_retry_options(..)	PulsarBroker::operation_retries
client.consumer().with_topic(..).with_subscription(..) и цикл по консьюмеру	обработчик #[subscriber] на PulsarSubscription
.with_subscription_type(..)	subscription_type в дескрипторе
.with_topics([..]), .with_topic_regex(..)	PulsarSubscription::topics, PulsarSubscription::pattern
десериализация msg.payload.data	тип параметра обработчика, который декодирует кодек
consumer.ack(&msg), consumer.nack(&msg)	исход обработчика (Подтверждение)
.with_dead_letter_policy(..)	max_attempts(n) и dead_letter(topic) в точке монтирования (Повторы и dead-letter)
.with_unacked_message_resend_delay(..)	ack_timeout в дескрипторе
consumer.seek(..)	start_at в точке монтирования или ручка перемотки в обработчике (Повторное чтение и перемотка)
client.producer().with_topic(..).build() и send_non_blocking	ответ или параметр Out, отправленный через политику публикации (Публикация и ответы)
producer::Message { partition_key, properties, .. }	шаг partition_key; заголовки становятся свойствами