Claude Code for RabbitMQ: Message Queues and Work Queues

RabbitMQ is the battle-tested message broker for work queues and pub/sub — amqp.connect(url) from amqplib creates a connection. conn.createChannel() opens a channel. channel.assertQueue(name, { durable: true }) creates a persistent queue. channel.sendToQueue(queue, Buffer.from(JSON.stringify(msg)), { persistent: true }) publishes a persistent message. Consumer: channel.consume(queue, handler, { noAck: false }), call channel.ack(msg) on success or channel.nack(msg, false, requeue) on failure. channel.prefetch(n) limits un-acked messages per consumer — enables fair work distribution. Exchange patterns: fanout broadcasts to all bound queues; topic routes by routingKey pattern (orders.#, orders.created); direct routes by exact key. Dead letter exchange: channel.assertQueue("jobs", { durable: true, arguments: { "x-dead-letter-exchange": "dlx", "x-dead-letter-routing-key": "failed" } }). Publisher confirms: await channel.confirmSelect(), then channel.publish(...) returns false if buffer is full — channel.waitForConfirms() resolves when broker confirms all. RPC: set replyTo: "amq.rabbitmq.reply-to" and a unique correlationId, then consume the reply queue. tls: { ca, cert, key } for TLS connections. Claude Code generates RabbitMQ work queues, topic routing, DLX patterns, and reliable job processors.

CLAUDE.md for RabbitMQ (amqplib)

## RabbitMQ Stack
- Version: amqplib >= 0.10
- Connect: const conn = await amqp.connect(process.env.RABBITMQ_URL!); const ch = await conn.createChannel()
- Queue: await ch.assertQueue("jobs", { durable: true }) — durable survives broker restart
- Publish: ch.sendToQueue("jobs", Buffer.from(JSON.stringify(payload)), { persistent: true })
- Consume: ch.consume("jobs", msg => { if (!msg) return; process(msg); ch.ack(msg) }, { noAck: false })
- Prefetch: ch.prefetch(10) — at most 10 unacknowledged messages per consumer
- Exchange: ch.assertExchange("events", "topic"); ch.bindQueue("orders-queue", "events", "orders.#")
- DLX: assertQueue("jobs", { arguments: { "x-dead-letter-exchange": "dlx" } })

RabbitMQ Connection Manager

// lib/rabbitmq/connection.ts — connection with auto-reconnect
import amqp, { type Connection, type Channel, type Options } from "amqplib"

type ManagedConnection = {
  connection: Connection
  channel:    Channel
}

let managed: ManagedConnection | null = null

async function createConnection(): Promise<ManagedConnection> {
  const url = process.env.RABBITMQ_URL ?? "amqp://localhost"
  const connection = await amqp.connect(url)
  const channel = await connection.createChannel()

  connection.on("error", (err) => {
    console.error("[RabbitMQ] Connection error:", err.message)
    managed = null
  })

  connection.on("close", () => {
    console.warn("[RabbitMQ] Connection closed — will reconnect on next use")
    managed = null
  })

  return { connection, channel }
}

export async function getChannel(): Promise<Channel> {
  if (!managed) {
    managed = await createConnection()
  }
  return managed.channel
}

export async function closeConnection(): Promise<void> {
  if (managed) {
    await managed.connection.close()
    managed = null
  }
}

process.once("SIGTERM", closeConnection)
process.once("SIGINT",  closeConnection)

Publisher

// lib/rabbitmq/publisher.ts — typed reliable publisher
import { getChannel } from "./connection"

export type JobPayload<T> = {
  jobType:    string
  jobId:      string
  createdAt:  string
  payload:    T
  retryCount?: number
}

const EXCHANGE    = "app-events"
const JOBS_QUEUE  = "jobs"
const DELAY_QUEUE = "jobs-delayed"

export async function setupExchanges(): Promise<void> {
  const ch = await getChannel()

  // Topic exchange for pub/sub events
  await ch.assertExchange(EXCHANGE, "topic", { durable: true })

  // Dead letter exchange
  await ch.assertExchange("dlx", "direct", { durable: true })
  await ch.assertQueue("jobs-failed", { durable: true })
  await ch.bindQueue("jobs-failed", "dlx", "failed")

  // Main work queue with DLX
  await ch.assertQueue(JOBS_QUEUE, {
    durable: true,
    arguments: {
      "x-dead-letter-exchange":    "dlx",
      "x-dead-letter-routing-key": "failed",
      "x-message-ttl":             30 * 60 * 1000,  // 30 min max job time
    },
  })
}

/** Publish a job to the work queue */
export async function publishJob<T>(
  jobType: string,
  payload: T,
  options: { priority?: number; expiration?: number } = {},
): Promise<string> {
  const ch = await getChannel()
  const jobId = `${jobType}-${Date.now()}-${Math.random().toString(36).slice(2)}`

  const job: JobPayload<T> = {
    jobType,
    jobId,
    createdAt: new Date().toISOString(),
    payload,
    retryCount: 0,
  }

  const publishOptions: Options.Publish = {
    persistent:  true,
    contentType: "application/json",
    ...(options.priority   ? { priority:   options.priority }                        : {}),
    ...(options.expiration ? { expiration: String(options.expiration) }              : {}),
  }

  ch.sendToQueue(JOBS_QUEUE, Buffer.from(JSON.stringify(job)), publishOptions)
  return jobId
}

/** Publish an event to the topic exchange */
export async function publishEvent<T>(
  routingKey: string,
  payload:    T,
): Promise<void> {
  const ch = await getChannel()
  ch.publish(
    EXCHANGE,
    routingKey,
    Buffer.from(JSON.stringify({ routingKey, payload, timestamp: Date.now() })),
    { persistent: true, contentType: "application/json" },
  )
}

Worker Consumer

// lib/rabbitmq/worker.ts — reliable job consumer with retry and DLQ
import { getChannel } from "./connection"
import type { JobPayload } from "./publisher"
import type { ConsumeMessage } from "amqplib"

const JOBS_QUEUE = "jobs"
const MAX_RETRIES = 3

export type JobHandler<T> = (payload: T, meta: { jobId: string; attempt: number }) => Promise<void>

type HandlerMap = Record<string, JobHandler<unknown>>

export async function startWorker(
  handlers:    HandlerMap,
  concurrency = 5,
): Promise<() => Promise<void>> {
  const ch = await getChannel()
  ch.prefetch(concurrency)

  await ch.consume(JOBS_QUEUE, async (msg: ConsumeMessage | null) => {
    if (!msg) return

    let job: JobPayload<unknown>
    try {
      job = JSON.parse(msg.content.toString()) as JobPayload<unknown>
    } catch {
      console.error("[Worker] Malformed message — rejecting without requeue")
      ch.nack(msg, false, false)
      return
    }

    const handler = handlers[job.jobType]

    if (!handler) {
      console.warn(`[Worker] No handler for jobType: ${job.jobType}`)
      ch.nack(msg, false, false)  // DLQ the unknown job type
      return
    }

    const attempt = (job.retryCount ?? 0) + 1

    try {
      await handler(job.payload, { jobId: job.jobId, attempt })
      ch.ack(msg)
    } catch (err) {
      console.error(`[Worker] Job ${job.jobId} failed (attempt ${attempt}):`, (err as Error).message)

      if (attempt >= MAX_RETRIES) {
        console.error(`[Worker] Job ${job.jobId} exhausted retries — sending to DLQ`)
        ch.nack(msg, false, false)  // Will be routed to DLX → jobs-failed
      } else {
        // Requeue with incremented retry count — small delay via setTimeout
        ch.nack(msg, false, false)  // Reject, let DLX handle or requeue separately
        await new Promise((r) => setTimeout(r, 1000 * 2 ** attempt))
        const { publishJob } = await import("./publisher")
        await publishJob(job.jobType, job.payload, {})
      }
    }
  })

  const stop = async () => {
    await ch.close()
    await closeConnection()
  }

  const { closeConnection } = await import("./connection")
  process.once("SIGTERM", stop)
  process.once("SIGINT",  stop)

  return stop
}

Next.js API + Worker

// app/api/emails/route.ts — enqueue email job via RabbitMQ
import { NextResponse } from "next/server"
import { z } from "zod"
import { publishJob } from "@/lib/rabbitmq/publisher"
import { auth } from "@/lib/auth"

const EmailSchema = z.object({
  to:       z.string().email(),
  template: z.string(),
  data:     z.record(z.unknown()),
})

export async function POST(req: Request) {
  const session = await auth()
  if (!session) return NextResponse.json({ error: "Unauthorized" }, { status: 401 })

  const body = EmailSchema.parse(await req.json())
  const jobId = await publishJob("send-email", { ...body, userId: session.user.id })

  return NextResponse.json({ jobId }, { status: 202 })
}

// worker/index.ts — long-running worker process
// import { startWorker } from "@/lib/rabbitmq/worker"
// import { setupExchanges } from "@/lib/rabbitmq/publisher"
// 
// async function main() {
//   await setupExchanges()
//   await startWorker({
//     "send-email": async ({ to, template, data }) => {
//       await sendEmail({ to, template, data })
//     },
//     "generate-report": async ({ userId, type }) => {
//       await generateReport(userId, type)
//     },
//   }, 10)
//   console.log("[Worker] Listening for jobs...")
// }
// main().catch(console.error)

For the Kafka alternative when needing durable event streaming with partitioned topics, ordered message delivery within a partition, consumer group replay from any offset, and horizontal scaling to billions of events — Kafka is the standard for high-throughput durable event logs while RabbitMQ is optimized for work queues with flexible routing (topic, fanout, headers), priority queues, and low-latency task dispatch, see the Kafka guide. For the BullMQ/Redis alternative when wanting a simpler queue implementation backed by Redis (no separate broker), job rate limiting, delayed jobs, repeatable jobs, and a built-in admin dashboard with Bull Board — BullMQ is easier to operate for Node.js-centric stacks while RabbitMQ provides more robust protocol-level routing, federation, and cross-language support, see the BullMQ guide. The Claude Skills 360 bundle includes RabbitMQ skill sets covering work queues, topic routing, and reliable job processors. Start with the free tier to try message queue generation.