1. Home
  2. Destinations
  3. RabbitMQ
DESTINATION · QUEUES & STREAMS

Postgres changes into RabbitMQ.

Persistent, confirmed publishes to an exchange with the row's ordering key as routing key — or straight to a queue.

The RabbitMQ sink publishes each change persistently and waits for the broker's confirm. By default the routing key is the row's ordering key (schema.table:pk), which lets topic exchanges route per table; set a fixed routing key to publish directly to a queue through the default exchange.

How delivery works

AspectBehaviour for RabbitMQ
MessageBody = the change JSON, content_type: application/json, persistent, message_id = idempotency key.
RoutingRouting key = configured value, or the ordering key when empty.
BatchingAll publishes of a batch, then every confirm is awaited; a nack fails the batch.
OrderingPublishes are in sequence order on one channel.
IdempotencyDedupe on message_id.
On failureA failed request is retried up to 5 times with exponential backoff (30 s, 1, 2, 4 min, with jitter). After the fifth failure the message is parked as undelivered, the console shows it with the last error, and one click replays it — later changes to the same row wait behind it so order is preserved.
Connection checkDials, opens a channel and, when an exchange is named, declares it passively.

Set up in three steps

  1. Connect your database

    Paste a connection string and press Test. Waltail checks the version (12+), that wal_level is logical, that the user may replicate, and that a slot is free — and shows the fix for anything that fails. Then it creates its own publication and replication slot. Connection guide →

  2. Add RabbitMQ as a destination

    Pick the tables to stream, choose RabbitMQ, and fill in Connection URL. Press Test — Waltail checks it can reach and write to RabbitMQ before anything is saved.

  3. Change a row

    Insert or update a row. The console shows the first event as it is captured and delivered; from then on, every committed change follows within about a second. Have rows that already exist? Run a backfill — confirm the estimate and they arrive through RabbitMQ as read events.

Configuration

Fields as the console asks for them. Secrets are encrypted at rest and never shown again.

FieldConsole labelRequiredNotes
urlConnection URLYesamqps://user:pass@host/vhost.
exchangeExchangeOptionalEmpty publishes through the default exchange.
routing_keyRouting keyOptionalOverrides the per-message ordering key; set a queue name when using the default exchange.

Every sink also has two tuning settings: batch size (messages per request, default 100) and rate limit (messages per second, default unlimited). Changes to one row are never in flight twice at once.

Example

Every change is this JSON envelope — data is the row after the change, old the row before it when replica identity provides it, null otherwise:

{
  "op": "update",
  "table": "public.orders",
  "data": { "id": 42, "status": "shipped", "total": 19.99 },
  "old":  { "id": 42, "status": "paid",    "total": 19.99 },
  "seq": 137,
  "idempotency_key": "9f1c3d2e-6b4a-4c7e-9d0f-2a1b3c4d5e6f",
  "commit_lsn": "0/1A2B3C8"
}

Message format reference →

Questions

How do I route by table?

Leave the routing key empty and bind queues on a topic exchange with patterns like public.orders:* — keys look like public.orders:42.

Does it work with CloudAMQP or Amazon MQ?

Yes — use the amqps:// URL they give you.

Related

NATS JetStream

Nats-Msg-Id dedupe

Apache Kafka

Any Kafka-protocol broker

Amazon SQS

Standard and FIFO queues

Use cases: Webhooks for your database · Cache invalidation · Search index sync · Event-driven services · Audit log and archive.

Start streaming to RabbitMQ.