Skip to main content

Topics and subscriptions

This guide reads one topic from several parts of a service, through subscriptions the service creates and subscriptions operations own.

A publisher sends to a topic. A subscription is a resource of its own, attached to one topic, and it keeps every message published to the topic after it was created. Each subscription of a topic receives every message. The consumers of one subscription share its messages. A service always reads a subscription.

The example is an orders service: billing, shipping, an audit and a report each read the topic orders.

One subscription per consumer​

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

bill and ship each have a subscription of their own on orders, so each receives every order. A new consumer of the topic is a new subscription; nothing changes for the publisher. A subscription receives only what is published after it exists, so a consumer added later does not see the orders before it.

GooglePubSub names the subscription by its short id in the broker's project, or by its full resource name, projects/{project}/subscriptions/{id}. create_with_topic makes the service own its topology: at startup it creates the topic and the subscription where they are missing, and leaves them alone where they exist. A subscription it creates has message ordering on, which Ordering keys uses.

Subscriptions operations own​

In production the topology is often infrastructure, created by Terraform or gcloud before the service runs. The service then only names what it reads.

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

#[subscriber("orders-audit")] reads the subscription orders-audit as it is. Nothing is created at startup, and a subscription that is not there fails on its stream with PubSubError::Receive. #[subscriber(GooglePubSub)] fixes the kind and leaves the name open. The mount site names the subscription with .name(..), so the deployment decides it, here from REPORTS_SUBSCRIPTION. In the manual form the descriptor is built where the handler is mounted, so it takes the name directly.

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

Against the emulator, create the two subscriptions before the service starts, the way operations would:

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

The topic orders exists once the service has started at least once, because billing and shipping create it.

Several replicas of the service​

Every replica of the service opens a streaming pull on the same subscriptions. Pub/Sub hands each message of a subscription to one of them, so the replicas share the work. Scaling out is starting more replicas; no setting changes. A message one replica gives back with retry may come back to another.

Inside one replica, a subscription handles one message at a time. workers(n) on the #[subscriber] attribute handles up to n at once; Ordering keys shows the keyed form.

Testing​

The tests in the example file publish an order to orders and check that billing and shipping both handled it. In process, a subscription the service only names reads the topic of its own name, so the test of orders-audit and orders-reports publishes there. Testing a Pub/Sub service covers what the in-process Pub/Sub models.