How to Add Pub/Sub to an Actix Web App

Actix Web handles routing, extractors and the HTTP server, and when one request has to trigger several pieces of follow-up work, such as reserving stock and sending a confirmation after an order, Pub/Sub keeps that work out of the request. The handler publishes an event once and returns 202, and each consumer gets its own copy with its own retries.

This guide adds Pub/Sub to an Actix Web 4 app with Encore, RabbitMQ with lapin, NATS with async-nats and SNS with SQS. The example publishes an OrderPlaced event that two subscribers handle: one reserves stock in Postgres, the other sends a confirmation. For a single queue of jobs handed to workers, see background jobs in Actix Web.

What you need to decide:

• Which broker carries the events and how it runs locally.
• How several consumers each get every event.
• How a consumer handles a message delivered twice.
• Where failed messages go when retries run out.

Pub/Sub options for Actix Web at a glance

Encore: Pub/Sub for your existing Actix Web app

Encore combines an open source infrastructure SDK for TypeScript, Go, Python and Rust with a platform that deploys into your own AWS or GCP account. The Rust SDK adds infrastructure to the Actix Web app you already have: you keep HttpServer, App and your handlers, and declare topics and subscriptions as statics. Under encore run they run locally. On deploy they become SNS and SQS on AWS or Pub/Sub on GCP.

The example uses Actix Web 4 and declares a database, a topic and two subscriptions:

use actix_web::{post, web, App, HttpResponse, HttpServer, Responder};
use encore::pubsub::{Message, Subscription, Topic};
use encore::storage::postgres::Database;
use serde::{Deserialize, Serialize};

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

#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderPlaced {
    order_id: String,
    sku: String,
    quantity: i32,
    email: String,
}

// One topic, any number of subscriptions. SNS and SQS on AWS, Pub/Sub on GCP.
static ORDER_PLACED: Topic<OrderPlaced> = Topic::new("order-placed");

// Each subscription gets its own copy of every message.
#[encore::pubsub::register]
static RESERVE_STOCK: Subscription<OrderPlaced> =
    Subscription::new(&ORDER_PLACED, "reserve-stock", reserve_stock).max_retries(10);

#[encore::pubsub::register]
static SEND_CONFIRMATION: Subscription<OrderPlaced> =
    Subscription::new(&ORDER_PLACED, "send-confirmation", send_confirmation).max_retries(5);

The stock handler records the order ID and updates stock in one transaction, so a redelivered message finds the marker row and changes nothing. A handler that returns an error is retried up to the subscription's limit:

async fn reserve_stock(msg: Message<OrderPlaced>) -> Result<(), encore::Error> {
    let order = &msg.data;
    let tx = DB.transaction().await?;
    let first_time = tx
        .execute(
            "INSERT INTO handled_events (subscription, order_id) VALUES ('reserve-stock', $1) \
             ON CONFLICT DO NOTHING",
            &[&order.order_id],
        )
        .await?;
    if first_time == 0 {
        return tx.rollback().await;
    }
    let reserved = tx
        .execute(
            "UPDATE stock SET available = available - $2 WHERE sku = $1 AND available >= $2",
            &[&order.sku, &order.quantity],
        )
        .await?;
    tx.commit().await?;
    if reserved == 0 {
        println!("not enough {} in stock for order {}", order.sku, order.order_id);
        return Ok(());
    }
    println!("reserved {} x {} for order {}", order.quantity, order.sku, order.order_id);
    Ok(())
}

async fn send_confirmation(msg: Message<OrderPlaced>) -> Result<(), encore::Error> {
    let order = &msg.data;
    // Pass the order ID as the email provider's idempotency key, so a retry
    // after a timeout does not send a second email.
    println!(
        "confirmation for order {} to {} (attempt {})",
        order.order_id, order.email, msg.delivery_attempt
    );
    Ok(())
}

The AND available >= $2 condition stops stock from going below zero. When it matches no row, the example only logs the shortfall; a real app would record the order as rejected or backordered, or publish an event for it, so the customer hears about it.

The Actix handler publishes and returns 202 with the message ID. encore::initialize() starts the subscriptions next to the server, and a small GET /stock/{sku} handler (not shown) reads the remaining stock:

#[derive(Deserialize)]
struct NewOrder {
    order_id: String,
    sku: String,
    quantity: i32,
    email: String,
}

#[post("/orders")]
async fn place_order(body: web::Json<NewOrder>) -> impl Responder {
    let body = body.into_inner();
    if !(1..=100).contains(&body.quantity) {
        return HttpResponse::BadRequest().body("quantity must be between 1 and 100");
    }
    let event = OrderPlaced {
        order_id: body.order_id,
        sku: body.sku,
        quantity: body.quantity,
        email: body.email,
    };
    match ORDER_PLACED.publish(&event).await {
        Ok(id) => HttpResponse::Accepted().json(serde_json::json!({ "message_id": id })),
        Err(_) => HttpResponse::InternalServerError().finish(),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    encore::initialize();
    let addr = std::env::var("ENCORE_LISTEN_ADDR").unwrap_or("0.0.0.0:8080".into());
    HttpServer::new(|| App::new().service(place_order).service(stock))
        .bind(addr)?
        .run()
        .await
}

The migration creates stock, seeded with ten mugs, and handled_events with a primary key on (subscription, order_id). Publishing the same order twice reserves stock once:

On deploy, Encore creates the topic, one queue per subscription with a dead-letter queue for messages that run out of retries in your AWS account, or the Pub/Sub equivalents on GCP, and provisions an IAM role for the service. Traces connect the POST /orders request with each subscriber run, so a slow or failing consumer is visible next to the order that caused it.

Try Encore →

RabbitMQ with lapin

RabbitMQ gives each consumer its own queue bound to a fanout exchange. With lapin you create a channel at startup, store it in web::Data, and publish from the handler:

let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
let channel = conn.create_channel().await?;
channel
    .exchange_declare("order-placed".into(), ExchangeKind::Fanout, ExchangeDeclareOptions::default(), FieldTable::default())
    .await?;
// Enable publisher confirms on the channel.
channel.confirm_select(ConfirmSelectOptions::default()).await?;

// In the handler:
channel
    .basic_publish("order-placed".into(), "".into(), BasicPublishOptions::default(), &serde_json::to_vec(&event)?, BasicProperties::default())
    .await?
    .await?; // wait for the publisher confirm

Each consumer declares its queue, binds it to the exchange and runs in a tokio::spawn task, acking after the work commits and nacking on failure.

Good to know

Requeued messages go straight back to the queue, so a failing message can loop. A dead-letter exchange with a delivery limit stops that. You run the broker in Docker locally and as a cluster in production.

Try RabbitMQ →

NATS with async-nats

With core NATS, a subscriber that is down misses the event. For order events you want JetStream, where a stream stores messages and each consumer is a durable consumer with acknowledgements and redelivery. async-nats exposes both through async_nats::jetstream.

Good to know

NATS is light to run and fast, but stream retention, consumer limits and the cluster are yours to operate in production.

Try NATS →

SNS and SQS with the AWS SDK

On AWS without Encore, the handler publishes to an SNS topic with aws-sdk-sns, each consumer has an SQS queue subscribed to the topic, and a Tokio task long-polls with aws-sdk-sqs, deleting each message after it is handled. You create the topic, the queues, the redrive policies, the queue policies that let SNS deliver and the IAM permissions yourself, usually in Terraform.

Good to know

Messages arrive wrapped in an SNS envelope unless raw delivery is enabled, and the visibility timeout must be longer than the slowest handler. Local development needs LocalStack or a development account.

Try Amazon SQS →

How to choose

Pick Encore when you want topics and subscriptions declared next to your Actix handlers, running under encore run and deployed as SNS and SQS or GCP Pub/Sub in your own account with dead-letter queues included. Pick RabbitMQ when you already operate it and want routing beyond fan-out. Pick NATS JetStream for low-latency messaging on infrastructure you run. Pick SNS and SQS by hand when your platform team already manages them in Terraform.

Whichever you pick, make each subscriber idempotent, because every option here delivers at least once.

You can sign up for early access to Encore for Rust to try it with your Actix Web app.

Related guides

• How to add background jobs to Actix Web
• How to add Postgres to Actix Web
• How to deploy Actix Web to AWS
• How to add cron jobs to Actix Web
• How to add secrets to Axum