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

Это руководство устанавливает ruststream-gcp-pubsub, запускает эмулятор Pub/Sub и первый сервис на нём.

Установка

Cargo.toml

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

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

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

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

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

Клиент под ним - google-cloud-pubsub. Он написан на Rust, поэтому сборке не нужны системные библиотеки. Крейту нужен Rust 1.95 или новее.

Локальный эмулятор

Google поставляет эмулятор Pub/Sub в образе Cloud SDK:

compose.yaml

services:
  pubsub:
    image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
    command: gcloud beta emulators pubsub start --host-port=0.0.0.0:8085 --project=my-project
    ports:
      - "8085:8085"

docker compose up -d

Эмулятор слушает localhost:8085. Он запускается пустым и принимает любой идентификатор проекта.

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

Макросы

src/main.rs

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

#[subscriber(GooglePubSub::new("orders-workers").create_with_topic("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(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |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(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(
                subscriber(
                    GooglePubSub::new("orders-workers").create_with_topic("orders"),
                    TakeOrder,
                )
                .build(),
            );
        },
    )
}

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

take_order читает заказы в JSON из подписки orders-workers. Заказ с суммой он принимает, а заказ без суммы отбрасывает. Сервис на Pub/Sub всегда читает подписку, а не тему. GooglePubSub описывает подписку, а create_with_topic при запуске создаёт её на теме orders вместе с самой темой. PubSubBroker::new принимает идентификатор проекта и только запоминает его. emulator направляет брокер на локальный эмулятор, потому что клиент на Rust не читает PUBSUB_EMULATOR_HOST. Соединение открывается при запуске сервиса. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Опубликуйте заказ из другого терминала через REST API эмулятора:

curl -X POST localhost:8085/v1/projects/my-project/topics/orders:publish \
  -H 'content-type: application/json' \
  -d "{\"messages\": [{\"data\": \"$(printf '{"id":1,"amount_cents":1250}' | base64)\"}]}"

В REST API Pub/Sub передаёт тело сообщения в base64, поэтому заказ сначала кодируется. Сервис пишет в журнал order taken с номером заказа. Ctrl+C останавливает его.

Первый тест

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

cargo test

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

Переход с google-cloud-pubsub

Сервис на google-cloud-pubsub сам строит клиенты, читает сообщения в цикле и публикует вручную. Вот куда переходит каждая часть в RustStream:

google-cloud-pubsub	RustStream
Subscriber::builder() с with_endpoint и with_credentials	PubSubBroker::new с endpoint, credentials или emulator
subscriber.subscribe(name) и цикл по потоку сообщений	обработчик #[subscriber(..)]
set_max_outstanding_messages, set_max_lease_extension, set_max_lease	max_outstanding, ack_extension, max_lease на дескрипторе (Управление потоком и аренда)
create_topic и create_subscription на административных клиентах	create_with_topic (Темы и подписки)
serde_json::from_slice над данными сообщения	тип параметра обработчика, который декодирует кодек
handler.ack(), handler.nack()	исход обработчика: ack, retry, retry_after, drop (Подтверждение и переобработка)
Publisher::publish	ответ или параметр Out, отправленный через политику Publish (Публикация и ответы)
Message::set_ordering_key	шаг ordering_key (Ключи упорядочивания)