Топики и подписки

Это руководство называет топики Pulsar, читает их через долговечные подписки и охватывает один топик, фиксированный список или все топики, которые подходят под шаблон.

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

Имена топиков

Имя топика Pulsar состоит из четырёх частей: persistent://public/default/orders. Первая часть - режим хранения, затем идут арендатор и пространство имён, затем сам топик. Короткое имя вроде orders означает persistent://public/default/orders, и тройка tenant/namespace/topic тоже обозначает хранимый топик. public/default - пространство имён, с которого начинается любая установка Pulsar.

Хранимый (persistent) топик хранит сообщения, пока их не подтвердят все подписки. Нехранимый топик держит их в памяти и отдаёт каждое только консьюмерам, подключённым в этот момент, поэтому он подходит для данных, которые через мгновение теряют ценность. Pulsar создаёт топик при первом обращении к нему, поэтому сервис только называет топики, которые читает и в которые пишет.

Сервис проверяет каждое имя топика до подключения, и неправильное имя останавливает его при запуске. PulsarTopic строит имя из частей или разбирает имя, которое приходит извне кода.

Один топик, две подписки

Pulsar читает топик только через подписку. Подписка - это курсор, который хранит сервер. Она помнит, что подтвердили её консьюмеры, и продолжает существовать после остановки сервиса.

Макросы

src/main.rs

/// Ships every order. The subscription `fulfilment` is this service's cursor on `orders`.
#[subscriber(PulsarSubscription::new("orders", "fulfilment"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

/// Reads the same topic under a subscription of its own, so it sees every order too.
#[subscriber(PulsarSubscription::new("orders", "notifications"))]
async fn notify(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "customer notified");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Ships every order. The subscription `fulfilment` is this service's cursor on `orders`.
struct Fulfil;

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

/// Reads the same topic under a subscription of its own, so it sees every order too.
struct Notify;

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

PulsarSubscription::new называет топик и подписку. fulfil и notify читают orders через две подписки, поэтому каждый из них получает каждый заказ. Два экземпляра сервиса делят один курсор на подписку, поэтому они распределяют заказы между собой, а не отгружают каждый дважды. Называйте подписку по сервису, которому она принадлежит: другой сервис, который назовёт ту же подписку, будет делить с ним её непрочитанные сообщения.

Новая подписка начинается с конца своего топика и видит только то, что опубликовано после неё. Существующая продолжает с того места, где остановились её консьюмеры, в том числе после перезапуска. Курсор переставляет руководство Повторное чтение и перемотка.

Фиксированный список топиков

Макросы

src/main.rs

/// Totals the orders of two regions through one consumer.
#[subscriber(PulsarSubscription::topics(["orders-eu", "orders-us"], "regional-totals"))]
async fn tally(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.region, "order counted");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Totals the orders of two regions through one consumer.
struct Tally;

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

PulsarSubscription::topics читает несколько топиков через один консьюмер под одним именем подписки. Каждый топик хранит под этим именем собственный курсор. У всех топиков один тип полезной нагрузки.

Шаблон

Макросы

src/main.rs

/// Audits every regional topic, those created after startup included.
#[subscriber(PulsarSubscription::pattern("persistent://public/default/orders-.*", "audit"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.region, "order audited");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Audits every regional topic, those created after startup included.
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.region, "order audited");
        ready(Ok(()))
    }
}

PulsarSubscription::pattern читает каждый топик, полное имя которого подходит под регулярное выражение, поэтому выражение записывается полностью. Подписка читает то пространство имён, которое названо в начале выражения: persistent://acme/orders/.* читает acme/orders, а выражение без части tenant/namespace/ читает public/default. audit читает orders-eu и orders-us, а также любой топик, созданный позже, полное имя которого подходит под выражение. Клиент заново получает список топиков каждые 30 секунд. Новый топик читается с того обновления списка, которое его нашло, а то, что попало в него раньше, не доставляется. Сервис останавливается при запуске, если выражение не является корректным регулярным выражением или если арендатор либо пространство имён в нём сами записаны регулярным выражением.

Короткая форма

Макросы

src/main.rs

/// Restocks returned items, under the broker's default subscription.
#[subscriber("returns")]
async fn restock(item: &Return) -> HandlerOutcome {
    tracing::info!(item.order, "return restocked");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650").default_subscription("fulfilment"),
        |b| {
            b.include(fulfil);
            b.include(notify);
            b.include(tally);
            b.include(audit);
            b.include(restock);
        },
    )
}

Вручную

src/main.rs

/// Restocks returned items, under the broker's default subscription.
struct Restock;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650").default_subscription("fulfilment"),
        |b| {
            b.include(subscriber(PulsarSubscription::new("orders", "fulfilment"), Fulfil).build());
            b.include(
                subscriber(PulsarSubscription::new("orders", "notifications"), Notify).build(),
            );
            b.include(
                subscriber(
                    PulsarSubscription::topics(["orders-eu", "orders-us"], "regional-totals"),
                    Tally,
                )
                .build(),
            );
            b.include(
                subscriber(
                    PulsarSubscription::pattern("persistent://public/default/orders-.*", "audit"),
                    Audit,
                )
                .build(),
            );
            b.include(subscriber("returns", Restock).build());
        },
    )
}

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

#[subscriber("returns")] называет топик, но не подписку. Обработчик присоединяется к подписке брокера по умолчанию default_subscription, здесь fulfilment, как к подписке Shared. Короткое имя на брокере без подписки по умолчанию останавливает сервис при запуске, и ошибка называет эту настройку. Любой другой настройке подписки, например её типу или тайм-ауту подтверждения, нужен дескриптор.

Тестирование подписок

Тесты в файле примера запускают приложение внутри процесса. Каждая подписка на топик получает каждое сообщение, опубликованное в него, как на сервере. Тест шаблона начинается с остановленными часами и сдвигает их на 30 секунд вперёд через tb.advance, поэтому следующее обновление списка происходит сразу. Работу с TestApp описывает руководство Тестирование сервиса на Pulsar.