Темы и подписки

Это руководство читает одну тему из нескольких частей сервиса через подписки, которые создаёт сам сервис, и через подписки, которыми владеет эксплуатация.

Издатель отправляет сообщение в тему. Подписка - самостоятельный ресурс, привязанный к одной теме. Она хранит каждое сообщение, опубликованное в тему после её создания. Каждая подписка на тему получает каждое сообщение. Потребители одной подписки делят её сообщения между собой. Сервис всегда читает подписку.

Пример - сервис заказов: тему orders читают биллинг, доставка, аудит и отчёт.

Своя подписка у каждого потребителя

Макросы

src/main.rs

/// Billing's own subscription on `orders`: it receives every order.
#[subscriber(GooglePubSub::new("orders-billing").create_with_topic("orders"))]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.amount_cents, "order billed");
    HandlerOutcome::ack()
}

/// Shipping's own subscription on the same topic: it receives every order too.
#[subscriber(GooglePubSub::new("orders-shipping").create_with_topic("orders"))]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Bills an order; mounted on billing's own subscription, it receives every order.
struct Bill;

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

/// Ships an order; mounted on shipping's own subscription, it receives every order too.
struct Ship;

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

У bill и ship своя подписка на orders, поэтому каждый получает каждый заказ. Новый потребитель темы - это новая подписка. Для издателя ничего не меняется. Подписка получает только то, что опубликовано после её создания, поэтому потребитель, добавленный позже, не видит более ранних заказов.

GooglePubSub называет подписку коротким идентификатором в проекте брокера или полным именем ресурса projects/{project}/subscriptions/{id}. С create_with_topic топологией владеет сам сервис: при запуске он создаёт тему и подписку, если их нет, и не трогает существующие. Подписка, которую он создаёт, включает упорядочивание сообщений. Его использует руководство Ключи упорядочивания.

Подписки, которыми владеет эксплуатация

В промышленной среде топология часто относится к инфраструктуре: её создают Terraform или gcloud до запуска сервиса. Тогда сервис только называет то, что читает.

Макросы

src/main.rs

/// A subscription operations created on `orders`; the service only reads it.
#[subscriber("orders-audit")]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order audited");
    HandlerOutcome::ack()
}

/// A subscription whose name the deployment decides, mounted with `.name(..)` below.
#[subscriber(GooglePubSub)]
async fn report(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order reported");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Audits an order; mounted on a subscription operations created.
struct Audit;

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

/// Reports an order; mounted on the subscription the deployment names.
struct Report;

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

#[subscriber("orders-audit")] читает подписку orders-audit такой, какая она есть. При запуске ничего не создаётся, а поток отсутствующей подписки возвращает ошибку PubSubError::Receive. #[subscriber(GooglePubSub)] фиксирует вид подписки и оставляет имя открытым. Точка монтирования называет подписку через .name(..), поэтому имя выбирает развёртывание, здесь из REPORTS_SUBSCRIPTION. В ручной форме дескриптор строится там, где монтируется обработчик, поэтому принимает имя сразу.

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    let reports = env::var("REPORTS_SUBSCRIPTION").unwrap_or_else(|_| "orders-reports".to_owned());
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(bill);
            b.include(ship);
            b.include(audit);
            b.include(report.name(reports));
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    let reports = env::var("REPORTS_SUBSCRIPTION").unwrap_or_else(|_| "orders-reports".to_owned());
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(
                subscriber(
                    GooglePubSub::new("orders-billing").create_with_topic("orders"),
                    Bill,
                )
                .build(),
            );
            b.include(
                subscriber(
                    GooglePubSub::new("orders-shipping").create_with_topic("orders"),
                    Ship,
                )
                .build(),
            );
            b.include(subscriber("orders-audit", Audit).build());
            b.include(subscriber(GooglePubSub::new(reports), Report).build());
        },
    )
}

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

На эмуляторе создайте обе подписки до запуска сервиса, как это сделала бы эксплуатация:

for subscription in orders-audit orders-reports; do
  curl -X PUT localhost:8085/v1/projects/my-project/subscriptions/$subscription \
    -H 'content-type: application/json' \
    -d '{"topic": "projects/my-project/topics/orders"}'
done

Тема orders существует, если сервис запускался хотя бы раз, потому что её создают биллинг и доставка.

Несколько реплик сервиса

Каждая реплика сервиса открывает streaming pull на одних и тех же подписках. Pub/Sub передаёт каждое сообщение подписки одной из них, поэтому реплики делят работу. Чтобы масштабировать сервис, запустите больше реплик. Настройки не меняются. Сообщение, которое одна реплика вернула через retry, может прийти к другой.

Внутри одной реплики подписка обрабатывает по одному сообщению. workers(n) в атрибуте #[subscriber] обрабатывает до n сообщений одновременно. Форму с ключом показывает руководство Ключи упорядочивания.

Тестирование

Тесты в файле примера публикуют заказ в orders и проверяют, что его обработали и биллинг, и доставка. Внутри процесса подписка, которую сервис только называет, читает одноимённую тему, поэтому тест orders-audit и orders-reports публикует туда. Что моделирует внутрипроцессный Pub/Sub, описывает руководство Тестирование сервиса на Pub/Sub.