Skip to main content

Publishing and replies

This guide publishes to Pub/Sub topics in the three ways a service does: a reply to a message, a publish from inside a handler, and a message once the service is up.

Every publish goes through a publish policy. Publish, the prelude's name for PubSubPublish, publishes to a topic and waits for the service to accept the message. A topic that does not exist fails the publish: Pub/Sub keeps no message for a topic it does not have.

The example is an orders service that answers each order with a receipt, bills it, and announces that it is up.

A reply​

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 }
}

The reply clause makes the return value a message to publish. Where it goes is a property of its type: #[outgoing(name = "receipts")] sends every Receipt to the topic receipts. The receipt is published after the handler returns and before the order is acknowledged, so a receipt that fails to publish leaves the order unacknowledged, and it comes back.

A publish from a handler​

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()
        }
    }
}

An Out parameter is a publisher the handler calls itself. bill names the topic on the call with .to("invoices") and decides what a failed publish means for the order. Here it asks for the order again with retry, so the invoice is not lost. A handler may publish any number of messages, to any topics, before it settles.

The mount site​

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
            });
        },
    )
}

.out_reply(Publish::default()) gives the reply its policy, and .out(DefaultSlot, Publish::default()) gives the Out parameter its own. A policy holds settings for every message of its position: Publish::default().ordering_key(..) is one. The broker creates one client publisher per topic, on the first publish to it, and every later publish to that topic reuses it.

after_startup runs once every subscription is open. It receives a live publisher of the policy it names, here to announce the service on services. A hook that fails stops the startup, so a service that cannot announce itself does not run.

Running it​

The topics the service publishes to belong to whoever reads them, so they exist before the service starts. Against the emulator, create them through its REST API:

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

Without services, the announcement fails and the service does not start.

Testing​

The tests in the example file publish an order and assert on what reached receipts and invoices with published. A second test starts the app and finds the announcement on services. In process, every publish is recorded whether or not a subscription reads its topic. Testing a Pub/Sub service covers the harness.