Публикация и ответы

Это руководство публикует в темы Pub/Sub тремя способами, которыми это делает сервис: ответом на сообщение, публикацией из обработчика и сообщением после запуска сервиса.

Каждая публикация проходит через политику публикации. Publish, имя PubSubPublish в прелюдии, публикует в тему и ждёт, пока Pub/Sub примет сообщение. Публикация в несуществующую тему возвращает ошибку: Pub/Sub не хранит сообщения для темы, которой у него нет.

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

Ответ

Макросы

src/main.rs

/// The receipt of an order; its type names the topic it goes to.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

#[subscriber(GooglePubSub::new("orders-receipts").create_with_topic("orders"), reply)]
async fn receipt(order: &Order) -> Receipt {
    tracing::info!(order.id, "receipt sent");
    Receipt { order_id: order.id }
}

Вручную

src/main.rs

/// The receipt of an order; its type names the topic it goes to.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

struct ReceiptFor;

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

Аргумент reply делает возвращаемое значение сообщением для публикации. Куда оно уходит, определяет его тип: #[outgoing(name = "receipts")] отправляет каждый Receipt в тему receipts. Квитанция публикуется после возврата из обработчика и до подтверждения заказа. Если квитанцию опубликовать не удалось, заказ остаётся неподтверждённым и приходит снова.

Публикация из обработчика

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Invoice {
    order_id: u64,
    amount_cents: u64,
}

/// Publishes the invoice of an order to the topic the call names.
#[subscriber(GooglePubSub::new("orders-billing").create_with_topic("orders"))]
async fn bill(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let invoice = Invoice {
        order_id: order.id,
        amount_cents: order.amount_cents,
    };
    match out.message(&invoice).to("invoices").publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, error = %err, "invoice not published, order retried");
            HandlerOutcome::retry()
        }
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Invoice {
    order_id: u64,
    amount_cents: u64,
}

/// Publishes the invoice of an order to the topic the call names.
struct Bill;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Bill
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let invoice = Invoice {
            order_id: order.id,
            amount_cents: order.amount_cents,
        };
        let sent = outs
            .get(DefaultSlot)
            .message(&invoice)
            .to("invoices")
            .publish()
            .await;
        if let Err(err) = sent {
            tracing::warn!(order.id, error = %err, "invoice not published, order retried");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Параметр Out - издатель, которого обработчик вызывает сам. bill называет тему в вызове через .to("invoices") и сам решает, что неудачная публикация значит для заказа. Здесь он запрашивает заказ снова через retry, поэтому счёт не теряется. До завершения доставки обработчик может опубликовать сколько угодно сообщений в любые темы.

Точка монтирования

Макросы

src/main.rs

#[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(receipt).out_reply(Publish::default());
            b.include(bill).out(DefaultSlot, Publish::default()).build();
            b.after_startup(Publish::default(), async move |publisher| {
                publisher
                    .message(&Started {
                        service: "orders".to_owned(),
                    })
                    .publish()
                    .await
            });
        },
    )
}

Вручную

src/main.rs

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-receipts").create_with_topic("orders"),
                    ReceiptFor,
                )
                .reply()
                .build(),
            )
            .out_reply(Publish::default());
            b.include(
                subscriber(
                    GooglePubSub::new("orders-billing").create_with_topic("orders"),
                    Bill,
                )
                .build(),
            )
            .out(DefaultSlot, Publish::default())
            .build();
            b.after_startup(Publish::default(), async move |publisher| {
                publisher
                    .message(&Started {
                        service: "orders".to_owned(),
                    })
                    .publish()
                    .await
            });
        },
    )
}

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

.out_reply(Publish::default()) задаёт политику ответа, а .out(DefaultSlot, Publish::default()) - отдельную политику параметра Out. Политика хранит настройки для каждого сообщения своей позиции, например Publish::default().ordering_key(..). Брокер создаёт один клиентский издатель на тему при первой публикации в неё, и все следующие публикации в эту тему используют его.

after_startup выполняется один раз, когда открыты все подписки. Он получает работающего издателя той политики, которую называет, здесь чтобы сообщить о сервисе в services. Если хук возвращает ошибку, запуск прерывается, поэтому сервис, который не может сообщить о себе, не работает.

Запуск

Темы, в которые публикует сервис, принадлежат тем, кто их читает, поэтому они существуют до запуска сервиса. На эмуляторе создайте их через REST API:

for topic in receipts invoices services; do
  curl -X PUT localhost:8085/v1/projects/my-project/topics/$topic
done
cargo run -- run

Без services сообщение о запуске возвращает ошибку, и сервис не запускается.

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

Тесты в файле примера публикуют заказ и через published проверяют, что дошло до receipts и invoices. Второй тест запускает приложение и находит сообщение о запуске в services. На внутрипроцессном Pub/Sub каждая публикация записывается, читает её тему какая-либо подписка или нет. Работу с обвязкой описывает руководство Тестирование сервиса на Pub/Sub.