# Pub/Sub > Decoupling services and building asynchronous systems Publishers & Subscribers (Pub/Sub) let you build systems that communicate by broadcasting events asynchronously. This is a great way to decouple services for better reliability and responsiveness. Encore.rs lets you declare topics and subscriptions in code. Encore provisions them for each environment: a managed Pub/Sub service in the cloud, and a local broker under `encore run`, with no setup. Here is the whole flow: a topic, a subscription that handles its messages, and a publish. ```rust use encore::pubsub::{Message, Subscription, Topic}; use serde::{Deserialize, Serialize}; #[derive(Clone, Serialize, Deserialize)] pub struct SignupEvent { pub user_id: String, } pub static SIGNUPS: Topic = Topic::new("signups"); #[encore::pubsub::register] static WELCOME: Subscription = Subscription::new(&SIGNUPS, "send-welcome", send_welcome); async fn send_welcome(msg: Message) -> Result<(), encore::Error> { println!("welcome, {}", msg.data.user_id); Ok(()) } async fn signed_up(user_id: String) -> Result<(), encore::Error> { SIGNUPS.publish(&SignupEvent { user_id }).await?; Ok(()) } ``` ## Creating a topic The core of Pub/Sub is the **topic**, a named channel on which you publish messages. A topic is typed by the message it carries, and declared as a module-level `static`: ```rust use encore::pubsub::Topic; use serde::{Deserialize, Serialize}; #[derive(Clone, Serialize, Deserialize)] pub struct SignupEvent { pub user_id: String, } pub static SIGNUPS: Topic = Topic::new("signups"); ``` Encore reads the declaration from your source code to provision the topic, so: * It must be a module-level `static`, not a local variable or a value built inside a function. * The name must be a string literal: lowercase letters, digits, dashes and periods (`user.signed-up`), starting with a letter. ### At-least-once delivery By default, a topic delivers each message to each subscription _at least once_: if Encore believes a message wasn't handled, it delivers it again. **So every subscription handler should be [idempotent](https://en.wikipedia.org/wiki/Idempotence#Computer_science_meaning)**: handling a message twice should have the same effect as handling it once. Track what you've already done in a database, or make the action itself idempotent. ### Exactly-once delivery Declare a topic `.exactly_once()` to have the infrastructure minimize redelivery: ```rust static PAYMENTS: Topic = Topic::new("payments").exactly_once(); ``` A message may still be redelivered in rare cases, such as when the acknowledgement that it was handled is lost on the network, so handlers should still be idempotent where correctness matters. Exactly-once delivery doesn't deduplicate publishes either: publish the same message twice and it's delivered twice. Cloud providers limit the throughput of exactly-once topics: - AWS: 300 messages per second for the topic (see [AWS SQS Quotas](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/quotas-messages.html)). - GCP: at least 3,000 messages per second across all topics in the region (see [GCP Pub/Sub Quotas](https://cloud.google.com/pubsub/quotas#quotas)). `.exactly_once()` is part of the declaration: it takes the topic by value, so it can't be called on the `static` later. ### Ordered topics A topic delivers messages in no particular order, which lets them be handled in parallel. When the messages about one entity must be handled in the order they were published, declare an `OrderedTopic`. Its `publish` takes an ordering key: messages with the same key are delivered in the order they were published, and messages with different keys in no particular order. ```rust use encore::pubsub::OrderedTopic; #[derive(Serialize, Deserialize)] pub struct CartEvent { pub cart_id: u64, pub event: String, } pub static CART_EVENTS: OrderedTopic = OrderedTopic::new("cart-events"); async fn checkout(cart_id: u64) -> Result<(), encore::Error> { let key = cart_id.to_string(); // Delivered in this order, as they share a key. CART_EVENTS.publish(&CartEvent { cart_id, event: "checkout_started".into() }, &key).await?; CART_EVENTS.publish(&CartEvent { cart_id, event: "checkout_completed".into() }, &key).await?; Ok(()) } ``` A message with a given key isn't delivered until the one before it has been handled or given up on, so one message that keeps failing holds up the rest of its key ([head-of-line blocking](https://en.wikipedia.org/wiki/Head-of-line_blocking)). Log and alert on failures, and choose the subscription's retry policy with that in mind. Cloud providers limit the throughput of ordered topics: - AWS: 300 messages per second for the topic (see [AWS SQS Quotas](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/quotas-messages.html)). - GCP: 1 MBps for each ordering key (see [GCP Pub/Sub Resource Limits](https://cloud.google.com/pubsub/quotas#resource_limits)). An ordered topic can also be `.exactly_once()`. ### Message encodings A topic's second type argument says how its messages are encoded: | Encoding | Message type | | | --- | --- | --- | | `Json` (the default) | Any type that implements serde's `Serialize` and `Deserialize` | `Topic` | | `Raw` | `Vec`, passed through as it is | `Topic, Raw>` | | `Protobuf` | A [`prost`](https://docs.rs/prost) message type | `Topic` | ```rust use encore::pubsub::{Raw, Topic}; static BLOBS: Topic, Raw> = Topic::new("blobs"); ``` `Protobuf` needs the SDK's `protobuf` feature: ```toml -- Cargo.toml -- [dependencies] encore-dev = { path = "/path/to/encore-dev", features = ["protobuf"] } ``` The encoding is part of the topic's type, so publishers and subscribers can't disagree about it. ## Publishing messages Call `publish` on the topic. It returns the message's id once the message has been published: ```rust let id = SIGNUPS.publish(&SignupEvent { user_id: "ada".into() }).await?; ``` Once `publish` returns, every subscription to the topic will receive the message. Subscribers see the same id. To send attributes alongside a message, use `publish_with`: ```rust use encore::pubsub::PublishOptions; SIGNUPS .publish_with( &event, PublishOptions { attributes: [("source".to_string(), "import".to_string())].into(), }, ) .await?; ``` On an `OrderedTopic`, `publish` and `publish_with` take the ordering key after the message. Encore reads which services publish to which topics from the calls to `publish`, and grants each of them permission to. So a topic can only be used by calling its methods on the `static` directly: it can't be stored, returned, or passed to a function. If library code needs to publish, give it a function that does. ## Subscribing to messages A subscription delivers every message published to its topic to a handler. Declare it as a module-level `static` with `Subscription::new`, taking the topic, a subscription name, and the handler: ```rust use encore::pubsub::{Message, Subscription}; #[encore::pubsub::register] static WELCOME: Subscription = Subscription::new(&SIGNUPS, "send-welcome", send_welcome); async fn send_welcome(msg: Message) -> Result<(), encore::Error> { // Send the email… Ok(()) } ``` * The name must be a string literal, unique for the topic: lowercase letters, digits and dashes. * The handler is an `async fn` taking a `Message` and returning `Result<(), encore::Error>`. Its future must be `Send`. * The subscription runs in the service that declares it, which can be the topic's service or any other. Each subscription receives every message, independently of the topic's other subscriptions: if one handler is slow, its subscription builds up a backlog while the others keep up. ### The message A handler receives a `Message`, with these fields: | Field | Type | | | --- | --- | --- | | `id` | `String` | The message's id, unique within its topic. | | `data` | `T` | The message, decoded. | | `raw_data` | `Vec` | The message as it was sent, before decoding. | | `attributes` | `HashMap` | The attributes it was published with. | | `publish_time` | `SystemTime` | When it was published. | | `delivery_attempt` | `u32` | Which attempt at delivering it this is, from 1. | | `ordering_key` | `Option` | The key it was published with, on an `OrderedTopic`. | ### Starting subscriptions A `static` does nothing until something uses it, so something has to start a subscription. `#[encore::pubsub::register]` registers one to start with the app. Registered subscriptions start when your `main` calls `encore::initialize()`: ```rust #[tokio::main] async fn main() { // Connects to Encore's runtime and starts the registered subscriptions. encore::initialize(); // Serve HTTP with your framework of choice… } ``` A service that only handles messages, and serves no requests of its own, can call `encore::run_blocking()` from a plain (non-async) `main` instead. It starts the registered subscriptions and blocks until the process is told to shut down. ```rust fn main() { encore::run_blocking(); } ``` A subscription declared without the attribute is started by hand with `WELCOME.start()`. Without the attribute, the handler must be written as a closure that boxes the future: `|msg| Box::pin(send_welcome(msg))`. The attribute is what lets you give the plain name of an `async fn`. If a service declares a subscription and never starts it, Encore logs a warning shortly after the service starts. When the process is told to shut down, Encore stops taking new messages and waits for in-flight handlers to finish. See [Graceful shutdown](/docs/rs/develop/shutdown). ### Subscription options Options follow the handler, each a method of its own: ```rust use std::time::Duration; #[encore::pubsub::register] static WELCOME: Subscription = Subscription::new(&SIGNUPS, "send-welcome", send_welcome) .ack_deadline(Duration::from_secs(60)) .max_retries(5) .max_concurrency(4); ``` | Option | Default | | | --- | --- | --- | | `.ack_deadline(Duration)` | 30 seconds | How long the handler has to finish before the message is delivered again. | | `.message_retention(Duration)` | 7 days | How long an undelivered message is kept. | | `.min_backoff(Duration)` | 10 seconds | The shortest wait before a failed message is delivered again. | | `.max_backoff(Duration)` | 10 minutes | The longest wait before a failed message is delivered again. | | `.max_retries(i32)` | 100 | How many times a failed message is retried before it's given up on. `-1` retries forever, `-2` never retries. | | `.max_concurrency(u32)` | Unlimited | How many messages are handled at once. | Encore reads the options from your source code to configure the infrastructure, so write them where the subscription is declared, with literal values: durations as `Duration::from_secs(…)` or `Duration::from_millis(…)`, and numbers as literals. Each option can be set once. ### Error handling If a handler returns an error, or a message can't be decoded as the topic's type, the message is delivered again later, with a backoff between `min_backoff` and `max_backoff`. After `max_retries` failed retries it's moved to the subscription's dead-letter queue, so the subscription can keep handling other messages until the bug is fixed. You can then release dead-lettered messages to be processed again. Locally, use the `encore pubsub` commands to see and manage what happened to each message: ```shell $ encore pubsub ls # topics, subscriptions and their queues $ encore pubsub queue user-added greet --state dead $ encore pubsub show # a message and its delivery attempts $ encore pubsub release user-added greet # deliver dead messages again ``` ## Using a topic from another service Any service can publish to a topic and subscribe to it. In a service other than the one that declares it, bind it by name with `Topic::named` (or `OrderedTopic::named`): ```rust use encore::pubsub::Topic; static SIGNUPS: Topic = Topic::named("signups"); ``` The name must be a string literal. A named topic takes no options: it's configured where it's declared, which may be a service in another language. Its message type and encoding are your claim: across languages Encore can't check them, so make sure they match what's published. ## Testing Under `encore test`, each test publishes to topics of its own, and no subscription runs: nothing handles what a test publishes. That lets you test publishers without the side effects of subscribers. To check what was published, `encore::testing::published` returns what the current test, and the code it called, published to a topic, in order. To test a handler, call it with what was published: ```rust #[cfg(test)] mod tests { use super::*; #[encore::test] async fn signing_up_sends_a_welcome() -> Result<(), encore::Error> { signed_up("ada".to_string()).await?; let published = encore::testing::published(&SIGNUPS)?; assert_eq!(published.len(), 1); assert_eq!(published[0].data.user_id, "ada"); // Run the subscription's handler on it. send_welcome(published[0].clone()).await?; Ok(()) } } ``` `Message` is `Clone` when `T` is. See [Testing](/docs/rs/develop/testing) for more. ## Ensuring consistency between services Database writes and publishes aren't transactional: a service can write to its database and then fail to publish, or the other way around. If that matters, consider the [transactional outbox pattern](https://microservices.io/patterns/data/transactional-outbox.html): write the message to an outbox table in the same transaction as the change, and publish from the table.