How to Add Pub/Sub to an Axum API

Axum handles routing, extractors and shared state, and when a handler needs to tell other parts of the system that something happened, such as sending a welcome email after a signup, the work moves out of the request and onto a message broker. Adding Pub/Sub means choosing a broker and a client crate, passing the client to handlers through State, running subscriber loops on the same Tokio runtime as the server, deciding how failed messages are retried and keeping a broker running locally and in tests. This guide goes through Encore, NATS, RabbitMQ, Kafka and Redis with working code and then looks at what changes in production.

What you need to decide:

• Which broker carries the messages and which crate talks to it.
• How the publisher reaches your handlers, which is usually Axum State.
• Where subscribers run and how they start and stop with the server.
• How failed messages are retried and where they end up when retries run out.
• Where the broker comes from locally, in tests and in production.

Pub/Sub options for Axum at a glance

Encore: add topics and subscriptions to your existing Axum app

Encore combines the topic, the subscriber and the broker into static declarations in the Axum crate you already have, added through an infrastructure SDK rather than a new framework, which Encore finds through static analysis and then runs locally under encore run and provisions on deploy as SNS and SQS in your AWS account or Pub/Sub in your GCP account. Routing stays in Axum, and handlers publish through the topic directly, so there is no client to construct at startup, no connection string to manage and no State to thread through the router.

A topic is typed by the message it carries, and a subscription names the topic and the async function that handles each message:

use axum::{extract::Path, http::StatusCode, routing::post, Router};
use encore::pubsub::{Message, Subscription, Topic};
use encore::storage::postgres::Database;
use serde::{Deserialize, Serialize};

static DB: Database = Database::new("users").migrations("./migrations");

/// Published when a user is added.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct UserAdded {
    name: String,
}

static USER_ADDED: Topic<UserAdded> = Topic::new("user-added");

// Started with the app by encore::initialize().
#[encore::pubsub::register]
static GREET: Subscription<UserAdded> = Subscription::new(&USER_ADDED, "greet", greet);

// Must be safe to retry: delivery is at least once, so the same user can
// arrive more than once. The welcomed table records who already got an email.
async fn greet(msg: Message<UserAdded>) -> Result<(), encore::Error> {
    let name = &msg.data.name;
    if DB
        .query_opt("SELECT 1 FROM welcomed WHERE name = $1", &[name])
        .await?
        .is_some()
    {
        return Ok(());
    }
    send_welcome_email(name).await?;
    DB.execute(
        "INSERT INTO welcomed (name) VALUES ($1) ON CONFLICT (name) DO NOTHING",
        &[name],
    )
    .await?;
    Ok(())
}

// Placeholder for your email provider.
async fn send_welcome_email(name: &str) -> Result<(), encore::Error> {
    println!("sending welcome email to {name}");
    Ok(())
}

async fn add_user(Path(name): Path<String>) -> StatusCode {
    // Idempotent, so a client that retries after a 500 does not add a second row.
    if DB
        .execute(
            "INSERT INTO users (name) VALUES ($1) ON CONFLICT (name) DO NOTHING",
            &[&name],
        )
        .await
        .is_err()
    {
        return StatusCode::INTERNAL_SERVER_ERROR;
    }
    // Published after the insert has committed.
    match USER_ADDED.publish(&UserAdded { name }).await {
        Ok(_) => StatusCode::CREATED,
        Err(_) => StatusCode::INTERNAL_SERVER_ERROR,
    }
}

#[tokio::main]
async fn main() {
    encore::initialize();
    let app = Router::new().route("/users/{name}", post(add_user));
    let addr = std::env::var("ENCORE_LISTEN_ADDR").unwrap_or("0.0.0.0:8080".into());
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    axum::serve(listener, app)
        .with_graceful_shutdown(encore::shutdown::signal())
        .await
        .unwrap();
    std::process::exit(encore::shutdown::drained().await.code());
}

The event is published only after the insert has committed, so a failed publish returns a 500 for a row that already exists. Because the insert ignores a duplicate name, the client can retry, and the retry publishes the event again, which is why the subscriber checks the welcomed table before sending. When every committed row must produce an event even if the process dies in between, write the event to an outbox table in the same transaction as the row and publish from that table instead.

Retries, backoff, the ack deadline and concurrency are options on the subscription declaration, so tuning them means updating the existing GREET declaration rather than adding a second subscription, which would run its own handler and send every welcome email twice. A handler that returns an error is delivered again with backoff until the retry limit is reached:

use std::time::Duration;

#[encore::pubsub::register]
static GREET: Subscription<UserAdded> = Subscription::new(&USER_ADDED, "greet", greet)
    .ack_deadline(Duration::from_secs(60))
    .min_backoff(Duration::from_secs(5))
    .max_retries(5)
    .max_concurrency(4);

When messages that share a key must be handled in the order they were published, an OrderedTopic takes the ordering key on publish:

use encore::pubsub::OrderedTopic;

static AUDIT: OrderedTopic<UserAdded> = OrderedTopic::new("audit");

async fn announce(name: &str) -> Result<(), encore::Error> {
    let event = UserAdded { name: name.to_string() };
    USER_ADDED.publish(&event).await?;
    AUDIT.publish(&event, name).await?;
    Ok(())
}

Under encore test no subscription is started and each test publishes to topics of its own, so a test reads back what the code under test published and calls the handler itself when it wants it run:

#[tokio::test]
async fn nothing_handles_a_message_until_the_test_does() {
    announce("ada").await.unwrap();

    let published = encore::testing::published(&USER_ADDED).unwrap();
    assert_eq!(published.len(), 1);
    assert_eq!(published[0].data.name, "ada");

    greet(published[0].clone()).await.unwrap();
}

Key features

• Topics and subscriptions are typed static declarations next to the code that publishes and handles the messages.
• An AI coding agent can add a new topic and its subscriber in one change and run them locally, and the deploy creates the SNS topic and SQS queue without a separate infrastructure change.
• encore run runs the topics and subscriptions locally, so there is no broker container or connection string.
• Subscriptions take retry, backoff, ack deadline, retention and concurrency options, and ordered topics deliver messages per key in order.
• Tests read back published messages with encore::testing::published without a broker.
• On deploy the same declarations become SNS and SQS with dead-letter queues in your AWS account, or Pub/Sub with dead-letter topics in your GCP account.

Good to know

Messages are JSON-encoded by default from Serde types, and Raw passes bytes through unchanged, so a topic shared with a non-Rust service needs both sides to agree on the format. The same SDK works with Actix Web and Rocket, which Axum vs Actix Web vs Rocket compares. Rust support is in beta, and you can sign up for access.

Try Encore →

NATS with async-nats: the lightweight route

NATS

NATS is a small, fast broker where publishers send to subjects and subscribers receive from them, and the async-nats crate is its official Tokio client.

use axum::{extract::{Path, State}, http::StatusCode, routing::post, Router};
use futures::StreamExt;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct UserAdded {
    name: String,
}

async fn add_user(
    State(nats): State<async_nats::Client>,
    Path(name): Path<String>,
) -> StatusCode {
    let payload = match serde_json::to_vec(&UserAdded { name }) {
        Ok(p) => p,
        Err(_) => return StatusCode::INTERNAL_SERVER_ERROR,
    };
    match nats.publish("user-added", payload.into()).await {
        Ok(()) => StatusCode::CREATED,
        Err(_) => StatusCode::INTERNAL_SERVER_ERROR,
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let url = std::env::var("NATS_URL").unwrap_or("nats://localhost:4222".into());
    let nats = async_nats::connect(url).await?;

    // The subscriber runs on the same runtime as the server.
    let mut sub = nats.subscribe("user-added").await?;
    tokio::spawn(async move {
        while let Some(msg) = sub.next().await {
            if let Ok(event) = serde_json::from_slice::<UserAdded>(&msg.payload) {
                println!("greeting {}", event.name);
            }
        }
    });

    let app = Router::new()
        .route("/users/{name}", post(add_user))
        .with_state(nats);
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

For local development you also need a broker running:

Key features

• async_nats::Client is cheap to clone, so it fits Axum State directly.
• Core NATS has very low latency, and JetStream adds persistence, consumers with acknowledgements and redelivery.
• Request and reply is built in for synchronous calls between services.

Good to know

Core NATS delivers at most once, so a subscriber that is down or crashes misses messages, and anything that must not be lost needs JetStream streams and consumers, which adds configuration for retention, acknowledgement and redelivery that you manage yourself.

Try NATS →

RabbitMQ with lapin: queues with acknowledgements

RabbitMQ

RabbitMQ routes messages from exchanges to queues, and the lapin crate is the common async AMQP client for Rust.

Key features

• Consumers acknowledge each message, and unacknowledged messages are redelivered.
• Dead-letter exchanges collect messages that are rejected or expire.
• Fanout and topic exchanges deliver one message to several queues.

Good to know

lapin exposes AMQP closely, so declaring exchanges, queues and bindings and handling channel errors and reconnects is code you write, typically with a connection pool such as deadpool-lapin shared through State.

Try RabbitMQ →

Kafka with rdkafka: a persistent event log

Kafka

Apache Kafka stores messages in partitioned, persistent logs that consumers read at their own offsets, and rdkafka wraps the librdkafka C library with an async FutureProducer and StreamConsumer.

Key features

• Messages are retained, so consumers can replay from an earlier offset.
• Partitions scale throughput and keep ordering per key.
• Consumer groups share partitions between instances.

Good to know

rdkafka builds librdkafka from C sources, which adds a C toolchain and CMake to your build, and Kafka has no per-message retry, so failed messages go to a retry or dead-letter topic that you implement.

Try Kafka →

Redis Pub/Sub: ephemeral notifications

Redis

Redis channels deliver each published message to whoever is subscribed at that moment, and the redis crate supports them over an async connection.

Key features

• It needs no extra infrastructure if you already run Redis.
• Latency is very low.

Good to know

Messages are not stored, so a subscriber that is offline or slow loses them, and work that must complete fits better on Redis Streams or one of the brokers above.

Try Redis →

Taking it to production

With NATS, RabbitMQ, Kafka or Redis, production means running the broker as a cluster or buying a hosted one, putting its URL and credentials in secrets, configuring networking so every service can reach it, creating streams, queues or topics and their dead-letter destinations, and monitoring consumer lag, and many teams describe that setup in Terraform.

With Encore, the topic and subscription declarations in the code are what get provisioned, so a deploy to AWS creates SNS topics and SQS queues with dead-letter queues and a deploy to GCP creates Pub/Sub topics and subscriptions with dead-letter topics, each with IAM permissions scoped to the services that publish and subscribe.

How to choose

• NATS for most self-managed setups that want a light broker.
• RabbitMQ for work queues with per-message acknowledgements and dead-lettering.
• Kafka for high-volume streams that need replay.
• Redis Pub/Sub for notifications where losing a message is acceptable.
• Encore if you want topics and subscriptions handled locally and in production from the same code.

Rust support in Encore is in beta, and you can sign up for access if you want to try it with your Axum application.

Related guides

• How to add a Postgres database to an Axum API
• Message queues vs Pub/Sub
• What is Pub/Sub?
• Best Rust web frameworks in 2026