Skip to content
actor-ts
English

AMQP (RabbitMQ)

AmqpActor integrates with RabbitMQ (and other AMQP 0.9.1 brokers). Supports exchanges (direct / topic / fanout / headers), queues, routing keys, ack/nack semantics.

import { ActorSystem, AmqpActor, AmqpOptions } from 'actor-ts';


const amqpOptions = AmqpOptions.create()
  .withUrl('amqp://guest:guest@rabbitmq:5672')
  // Queue → consumer-actor wiring lives in `bindings` on the
  // settings — declared once when the actor connects.  Runtime
  // subscribe / unsubscribe isn't a v1 surface; if you need to
  // attach a new consumer after start, restart the actor with
  // updated bindings.
  .withBindings([
    { queue: 'order-processor', exchange: 'orders', routingKey: 'placed.#', target: handler },
  ]);
const amqp = system.spawn(
  () => new AmqpActor(
    amqpOptions,
  ),
  'amqp',
);


// Publish to an exchange:
amqp.tell({
  kind: 'publish',
  publish: {
    exchange:   'orders',
    routingKey: 'placed.priority',
    content:    JSON.stringify(order),
  },
});

Settings

Section titled “Settings”
interface AmqpOptionsType extends BrokerCommonOptionsType {
  url:        string;                          // amqp:// or amqps://
  prefetch?:  number;                          // unacked messages per consumer; default 1
  autoAcknowledge?:      boolean;                         // ack on delivery vs after processing; default true
  bindings?:  ReadonlyArray<AmqpQueueBinding>; // queue ↔ exchange ↔ routingKey ↔ target
}


type AmqpQueueBinding = {
  queue:       string;
  exchange?:   string;                         // omit for the default exchange
  routingKey?: string;
  target:      ActorRef<AmqpDelivery>;         // consumer-actor for this queue
};

Bindings declare the queue, the exchange binding, AND the consumer-actor target in one go — the actor wires everything up on connect (idempotent re-declaration).

Exchanges and routing

Section titled “Exchanges and routing”
TypeRouting
directExact match on routing key.
topicWildcard pattern match (* = one segment, # = many).
fanoutAll bound queues, ignoring routing key.
headersMatch on message headers, not routing key.

AmqpActor does not declare exchanges — they must already exist on the broker (created via RabbitMQ configuration, the management UI, or your infrastructure tooling). On connect the actor only asserts the queues named in bindings and binds them to the given exchange; publishing then routes through that existing exchange by name.

Most common pattern: topic exchanges with hierarchical routing keys (orders.placed.priority, orders.cancelled, payments.failed).

Ack / nack

Section titled “Ack / nack”

Set autoAcknowledge: false in settings, then the handler tells the result back via { kind: 'acknowledgment', delivery } or { kind: 'negativeAcknowledgment', delivery, requeue } on the AMQP actor’s ref.

class OrderHandler extends Actor<AmqpDelivery> {
  constructor(private readonly amqp: ActorRef<AmqpCommand>) { super(); }


  override async onReceive(delivery: AmqpDelivery): Promise<void> {
    try {
      await processOrder(delivery);
      this.amqp.tell({ kind: 'acknowledgment', delivery });
    } catch (err) {
      this.amqp.tell({ kind: 'negativeAcknowledgment', delivery, requeue: true });
    }
  }
}

Every inbound AmqpDelivery carries an opaque ackToken the actor uses to identify the broker-side message — you don’t touch it directly, just pass the whole delivery back in the ack / nack tell.

The broker holds the message in unacked state until your handler responds — if the consumer crashes, the broker re-delivers on the next connection. Use requeue: false to send to a dead-letter queue (if configured) — useful when retry won’t help.

Prefetch

Section titled “Prefetch”
prefetch: 10   // consume up to 10 unacked at once per consumer

The framework default is 1 — a single unacked message per consumer at a time. (The raw AMQP protocol default is 0 = unlimited, but the actor sets prefetch to 1 unless you override it.) Raise it to a sensible batch size (10-50) to keep the consumer busy without overwhelming it.

Peer dependency

Section titled “Peer dependency”
Terminal window
npm install amqplib
# or: bun add amqplib

When to use AMQP

Section titled “When to use AMQP”

Three primary scenarios:

  1. Work queues — multiple consumers competing for messages on a queue, one consumer wins per message.
  2. Routing + filtering — topic exchanges with rich routing keys.
  3. Bridging existing RabbitMQ infrastructure.

For high-throughput streaming, Kafka scales better. For lightweight pub/sub, NATS or MQTT is lighter. AMQP shines for enterprise-style queue + routing where features (DLQ, TTL, prefetch, ack semantics) matter.

Where to next

Section titled “Where to next”