The Complete Guide to RabbitMQ: From Beginner to Hero

{
  // Content
  content: "Your actual data here",
  contentType: "application/json",
  contentEncoding: "utf-8",
  
  // Routing
  routingKey: "user.events.created",
  exchange: "user_events",
  
  // Delivery guarantees
  deliveryMode: 2, // 1=non-persistent, 2=persistent
  priority: 5, // 0-9, higher = more important
  
  // Timestamps
  timestamp: 1634567890123,
  expiration: "86400000", // Message TTL in milliseconds
  
  // Headers for custom metadata
  headers: {
    "source": "user-service",
    "version": "1.0",
    "trace-id": "abc123"
  },
  
  // Unique identifier
  messageId: "unique-message-id-123"
}

// Durable queue (survives broker restart)
queue: {
  name: "user_notifications",
  durable: true,
  autoDelete: false
}

// Temporary queue (deleted when last consumer disconnects)
queue: {
  name: "", // Random name generated by broker
  durable: false,
  autoDelete: true,
  exclusive: true // Only used by this connection
}

// Queue with TTL and limits
queue: {
  name: "processing_queue",
  messageTTL: 3600000, // 1 hour
  maxLength: 10000, // Max 10,000 messages
  overflow: "drop-head" // Drop oldest when full
}

// API Service
app.post('/process-video', async (req, res) => {
  const taskId = generateId()
  
  // Publish task to workers
  await channel.publish('tasks', 'video.process', {
    taskId,
    videoUrl: req.body.videoUrl,
    options: req.body.options
  })
  
  // Return immediately
  res.json({ taskId, status: 'queued' })
})

// Worker Service
await channel.consume('video_processing', async (msg) => {
  const task = JSON.parse(msg.content.toString())
  
  try {
    const result = await processVideo(task.videoUrl, task.options)
    
    // Publish result
    await channel.publish('results', `task.${task.taskId}`, {
      taskId: task.taskId,
      result,
      status: 'completed'
    })
    
    channel.ack(msg)
  } catch (error) {
    // Publish error
    await channel.publish('results', `task.${task.taskId}`, {
      taskId: task.taskId,
      error: error.message,
      status: 'failed'
    })
    
    // Reject with requeue
    channel.nack(msg, false, true)
  }
})

// User Service (Publisher)
async function createUser(userData) {
  const user = await db.users.create(userData)
  
  // Publish domain events
  await channel.publish('user.events', 'user.created', {
    userId: user.id,
    email: user.email,
    timestamp: new Date().toISOString(),
    source: 'user-service'
  })
  
  return user
}

// Email Service (Subscriber)
await channel.consume('email_notifications', async (msg) => {
  const event = JSON.parse(msg.content.toString())
  
  if (event.routingKey === 'user.created') {
    await sendWelcomeEmail(event.email)
  }
  
  channel.ack(msg)
})

// Analytics Service (Subscriber)
await channel.consume('analytics_events', async (msg) => {
  const event = JSON.parse(msg.content.toString())
  
  // Track user signup event
  await analytics.track('user_signup', {
    userId: event.userId,
    source: event.source,
    timestamp: event.timestamp
  })
  
  channel.ack(msg)
})

// RPC Client Helper
class RPCClient {
  constructor(channel, serviceName) {
    this.channel = channel
    this.serviceName = serviceName
    this.responseQueue = null
    this.pendingRequests = new Map()
  }
  
  async call(method, params) {
    const correlationId = generateId()
    const replyTo = `${this.serviceName}_response_${process.pid}`
    
    // Create temporary response queue
    if (!this.responseQueue) {
      this.responseQueue = await this.channel.assertQueue(replyTo, {
        exclusive: true,
        autoDelete: true
      })
      
      await this.channel.consume(this.responseQueue.queue, (msg) => {
        const correlationId = msg.properties.correlationId
        const resolver = this.pendingRequests.get(correlationId)
        
        if (resolver) {
          resolver(JSON.parse(msg.content.toString()))
          this.pendingRequests.delete(correlationId)
        }
      })
    }
    
    // Send RPC request
    return new Promise((resolve, reject) => {
      this.pendingRequests.set(correlationId, { resolve, reject })
      
      this.channel.publish('rpc', `${this.serviceName}.${method}`, JSON.stringify(params), {
        correlationId,
        replyTo,
        expiration: 30000 // 30 second timeout
      })
      
      // Timeout handling
      setTimeout(() => {
        if (this.pendingRequests.has(correlationId)) {
          this.pendingRequests.delete(correlationId)
          reject(new Error('RPC timeout'))
        }
      }, 30000)
    })
  }
}

// Usage in Order Service
const inventoryClient = new RPCClient(channel, 'inventory')
const paymentClient = new RPCClient(channel, 'payment')

async function createOrder(orderData) {
  try {
    // Check inventory
    const inventoryResult = await inventoryClient.call('checkStock', {
      productId: orderData.productId,
      quantity: orderData.quantity
    })
    
    if (!inventoryResult.available) {
      throw new Error('Out of stock')
    }
    
    // Process payment
    const paymentResult = await paymentClient.call('processPayment', {
      amount: orderData.amount,
      paymentMethod: orderData.paymentMethod
    })
    
    if (!paymentResult.success) {
      throw new Error('Payment failed')
    }
    
    // Create order
    return await db.orders.create(orderData)
  } catch (error) {
    console.error('Order creation failed:', error)
    throw error
  }
}

// Inventory Service (RPC Server)
await channel.consume('inventory', async (msg) => {
  const { method, params } = JSON.parse(msg.content.toString())
  const { replyTo, correlationId } = msg.properties
  
  try {
    let result
    
    switch (method) {
      case 'checkStock':
        result = await checkInventory(params.productId, params.quantity)
        break
      case 'reserveStock':
        result = await reserveStock(params.productId, params.quantity)
        break
      default:
        throw new Error(`Unknown method: ${method}`)
    }
    
    // Send response
    await channel.sendToQueue(replyTo, Buffer.from(JSON.stringify(result)), {
      correlationId
    })
    
    channel.ack(msg)
  } catch (error) {
    // Send error response
    await channel.sendToQueue(replyTo, Buffer.from(JSON.stringify({
      error: error.message
    })), {
      correlationId
    })
    
    channel.nack(msg)
  }
})

version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:3.12-management
    container_name: rabbitmq
    ports:
      - "5672:5672"   # AMQP port
      - "15672:15672"  # Management UI port
      - "15692:15692"  # STOMP port
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123
      RABBITMQ_DEFAULT_VHOST: "/"
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "ping"]
      interval: 30s
      timeout: 10s
      retries: 3

  # Optional: Management UI
  rabbitmq-ui:
    image: rabbitmq:3.12-management-alpine
    container_name: rabbitmq-ui
    ports:
      - "15673:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123
    depends_on:
      - rabbitmq

volumes:
  rabbitmq_data:

docker-compose up -d

# Ubuntu/Debian
sudo apt-get update
sudo apt-get install rabbitmq-server

# macOS with Homebrew
brew install rabbitmq

# Start service
sudo systemctl start rabbitmq-server  # Linux
brew services start rabbitmq        # macOS

npm install amqplib
# or
yarn add amqplib

# For TypeScript
npm install --save-dev @types/node

const amqp = require('amqplib')

const rabbitConfig = {
  url: process.env.RABBITMQ_URL || 'amqp://admin:admin123@localhost:5672',
  
  // Connection options
  socketOptions: {
    timeout: 30000, // 30 seconds
    heartbeat: 60     // Heartbeat every 60 seconds
  },
  
  // Client properties
  clientProperties: {
    product_name: 'my-app',
    product_version: '1.0.0',
    platform: 'Node.js',
    capabilities: {
      publisher_confirms: true,
      consumer_cancel_notify: true,
      basic.nack: true
    }
  },
  
  // Reconnection settings
  retry: {
    times: 10,
    interval: 1000
  }
}

// Connection helper
async function createConnection() {
  try {
    const connection = await amqp.connect(rabbitConfig.url, {
      timeout: rabbitConfig.socketOptions.timeout,
      heartbeat: rabbitConfig.socketOptions.heartbeat,
      clientProperties: rabbitConfig.clientProperties
    })
    
    console.log('Connected to RabbitMQ')
    return connection
  } catch (error) {
    console.error('Failed to connect to RabbitMQ:', error)
    throw error
  }
}

module.exports = {
  rabbitConfig,
  createConnection
}

const amqp = require('amqplib')
const { createConnection } = require('./rabbitmqConfig')

class EventProducer {
  constructor() {
    this.connection = null
    this.channel = null
  }
  
  async connect() {
    try {
      // Create connection and channel
      this.connection = await createConnection()
      this.channel = await this.connection.createChannel()
      
      // Enable publisher confirms
      await this.channel.confirmChannel()
      
      // Declare exchange
      await this.channel.assertExchange('user_events', 'topic', {
        durable: true
      })
      
      console.log('Producer connected and ready')
    } catch (error) {
      console.error('Failed to connect producer:', error)
      throw error
    }
  }
  
  async publishEvent(routingKey, event, options = {}) {
    if (!this.channel) {
      throw new Error('Producer not connected')
    }
    
    try {
      const message = {
        id: generateId(),
        type: event.type,
        data: event.data,
        timestamp: new Date().toISOString(),
        source: 'user-service',
        version: '1.0'
      }
      
      const published = this.channel.publish(
        'user_events',           // Exchange
        routingKey,             // Routing key
        Buffer.from(JSON.stringify(message)), // Message body
        {
          persistent: true,        // Survive broker restart
          messageId: message.id,  // Unique message ID
          timestamp: Date.now(),     // Message timestamp
          expiration: options.ttl || '86400000', // 24 hours TTL
          headers: {
            'content-type': 'application/json',
            'event-type': event.type,
            ...options.headers
          }
        }
      )
      
      if (published) {
        console.log(`Event published: ${routingKey}`, message)
        return message
      } else {
        throw new Error('Failed to publish event')
      }
    } catch (error) {
      console.error('Error publishing event:', error)
      throw error
    }
  }
  
  async publishUserEvent(eventType, userData, options = {}) {
    const routingKey = `user.${eventType}`
    const event = {
      type: eventType,
      data: userData
    }
    
    return await this.publishEvent(routingKey, event, options)
  }
  
  async disconnect() {
    try {
      if (this.channel) {
        await this.channel.close()
      }
      if (this.connection) {
        await this.connection.close()
      }
      console.log('Producer disconnected')
    } catch (error) {
      console.error('Error disconnecting producer:', error)
    }
  }
}

// Helper function
function generateId() {
  return Date.now().toString(36) + Math.random().toString(36).substr(2)
}

// Usage example
async function main() {
  const producer = new EventProducer()
  
  try {
    await producer.connect()
    
    // Publish different user events
    await producer.publishUserEvent('created', {
      userId: 'user123',
      email: 'john@example.com',
      name: 'John Doe'
    })
    
    await producer.publishUserEvent('updated', {
      userId: 'user123',
      changes: { email: 'john.doe@example.com' }
    })
    
    await producer.publishUserEvent('login', {
      userId: 'user123',
      ip: '192.168.1.100',
      userAgent: 'Mozilla/5.0...'
    }, {
      headers: {
        'priority': 'high',
        'source': 'web-app'
      }
    })
    
    // Wait for confirms
    await new Promise(resolve => setTimeout(resolve, 1000))
    
  } catch (error) {
    console.error('Error in main:', error)
  } finally {
    await producer.disconnect()
  }
}

main()

class ResilientProducer {
  constructor() {
    this.connection = null
    this.channel = null
    this.isConnecting = false
    this.reconnectAttempts = 0
    this.maxReconnectAttempts = 10
  }
  
  async connect() {
    if (this.isConnecting) return
    
    this.isConnecting = true
    
    while (this.reconnectAttempts < this.maxReconnectAttempts) {
      try {
        this.connection = await createConnection()
        this.channel = await this.connection.createChannel()
        
        // Set up error handlers
        this.connection.on('error', this.handleConnectionError.bind(this))
        this.connection.on('close', this.handleConnectionClose.bind(this))
        this.channel.on('error', this.handleChannelError.bind(this))
        
        // Enable publisher confirms
        await this.channel.confirmChannel()
        
        // Assert exchange
        await this.channel.assertExchange('events', 'topic', {
          durable: true
        })
        
        console.log('Producer connected successfully')
        this.reconnectAttempts = 0
        this.isConnecting = false
        return
        
      } catch (error) {
        this.reconnectAttempts++
        console.error(`Connection attempt ${this.reconnectAttempts} failed:`, error)
        
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
          // Exponential backoff
          const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts - 1), 30000)
          await new Promise(resolve => setTimeout(resolve, delay))
        }
      }
    }
    
    this.isConnecting = false
    throw new Error('Failed to connect after maximum attempts')
  }
  
  async publishWithRetry(routingKey, message, maxRetries = 3) {
    let lastError
    
    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        if (!this.channel) {
          await this.connect()
        }
        
        const result = await this.publishOnce(routingKey, message)
        console.log(`Message published on attempt ${attempt}`)
        return result
        
      } catch (error) {
        lastError = error
        console.error(`Publish attempt ${attempt} failed:`, error)
        
        if (attempt < maxRetries) {
          // Wait before retry
          await new Promise(resolve => setTimeout(resolve, 1000 * attempt))
        }
      }
    }
    
    throw lastError
  }
  
  async publishOnce(routingKey, message) {
    return new Promise((resolve, reject) => {
      const messageId = generateId()
      
      // Set up confirm handler
      const confirmHandler = (correlationId, ack) => {
        if (correlationId === messageId) {
          this.channel.removeListener('confirm', confirmHandler)
          this.channel.removeListener('return', returnHandler)
          
          if (ack) {
            resolve({ messageId, success: true })
          } else {
            reject(new Error('Message not acknowledged'))
          }
        }
      }
      
      // Set up return handler (message returned)
      const returnHandler = (correlationId, returned) => {
        if (correlationId === messageId) {
          this.channel.removeListener('confirm', confirmHandler)
          this.channel.removeListener('return', returnHandler)
          reject(new Error('Message returned: ' + returned.reason))
        }
      }
      
      this.channel.on('confirm', confirmHandler)
      this.channel.on('return', returnHandler)
      
      // Publish message
      const published = this.channel.publish(
        'events',
        routingKey,
        Buffer.from(JSON.stringify(message)),
        {
          messageId,
          persistent: true,
          timestamp: Date.now(),
          headers: {
            'content-type': 'application/json',
            'publish-attempt': Date.now().toString()
          }
        }
      )
      
      if (!published) {
        reject(new Error('Channel write buffer full'))
      }
    })
  }
  
  handleConnectionError(error) {
    console.error('Connection error:', error)
    // Attempt to reconnect
    this.connect().catch(console.error)
  }
  
  handleConnectionClose() {
    console.log('Connection closed, attempting to reconnect...')
    this.connect().catch(console.error)
  }
  
  handleChannelError(error) {
    console.error('Channel error:', error)
  }
}

const amqp = require('amqplib')
const { createConnection } = require('./rabbitmqConfig')

class EventConsumer {
  constructor(queueName, handler) {
    this.queueName = queueName
    this.handler = handler
    this.connection = null
    this.channel = null
    this.isProcessing = false
  }
  
  async start() {
    try {
      // Create connection and channel
      this.connection = await createConnection()
      this.channel = await this.connection.createChannel()
      
      // Set prefetch count (QoS)
      await this.channel.prefetch(1)
      
      // Declare exchange
      await this.channel.assertExchange('user_events', 'topic', {
        durable: true
      })
      
      // Declare queue
      await this.channel.assertQueue(this.queueName, {
        durable: true,
        arguments: {
          'x-message-ttl': 86400000, // 24 hours
          'x-dead-letter-exchange': 'user_events_dlx',
          'x-dead-letter-routing-key': this.queueName
        }
      })
      
      // Bind queue to exchange
      await this.channel.bindQueue(this.queueName, 'user_events', 'user.*')
      
      // Set up consumer
      await this.channel.consume(this.queueName, this.handleMessage.bind(this), {
        noAck: false // Manual acknowledgment
      })
      
      console.log(`Consumer started for queue: ${this.queueName}`)
      
      // Set up error handlers
      this.connection.on('error', this.handleConnectionError.bind(this))
      this.connection.on('close', this.handleConnectionClose.bind(this))
      this.channel.on('error', this.handleChannelError.bind(this))
      this.channel.on('cancel', this.handleConsumerCancel.bind(this))
      
    } catch (error) {
      console.error('Failed to start consumer:', error)
      throw error
    }
  }
  
  async handleMessage(message) {
    if (this.isProcessing) {
      console.log('Already processing, skipping message')
      return
    }
    
    this.isProcessing = true
    
    try {
      const event = JSON.parse(message.content.toString())
      
      console.log(`Received event: ${message.fields.routingKey}`, event)
      
      // Process the event
      await this.handler(event, message)
      
      // Acknowledge message
      this.channel.ack(message)
      console.log(`Message processed and acknowledged`)
      
    } catch (error) {
      console.error('Error processing message:', error)
      
      // Negative acknowledgment with requeue
      this.channel.nack(message, false, true)
      console.log(`Message rejected and requeued`)
      
    } finally {
      this.isProcessing = false
    }
  }
  
  async stop() {
    try {
      if (this.channel) {
        // Cancel consumer
        await this.channel.cancel(this.queueName)
        
        // Close channel
        await this.channel.close()
      }
      
      if (this.connection) {
        await this.connection.close()
      }
      
      console.log(`Consumer stopped for queue: ${this.queueName}`)
    } catch (error) {
      console.error('Error stopping consumer:', error)
    }
  }
  
  handleConnectionError(error) {
    console.error('Connection error:', error)
  }
  
  handleConnectionClose() {
    console.log('Connection closed')
  }
  
  handleChannelError(error) {
    console.error('Channel error:', error)
  }
  
  handleConsumerCancel() {
    console.log('Consumer cancelled by server')
  }
}

// Usage example: Email notification consumer
async function emailNotificationHandler(event, message) {
  switch (event.type) {
    case 'created':
      await sendWelcomeEmail(event.data.email, event.data.name)
      break
      
    case 'updated':
      await sendProfileUpdateEmail(event.data.email, event.data.changes)
      break
      
    case 'login':
      await sendLoginNotificationEmail(event.data.email, event.data.ip)
      break
      
    default:
      console.log(`Unknown event type: ${event.type}`)
  }
}

// Usage example: Analytics consumer
async function analyticsHandler(event, message) {
  // Track user events in analytics
  await analytics.track(event.type, {
    userId: event.data.userId,
    timestamp: event.timestamp,
    source: event.source,
    routingKey: message.fields.routingKey
  })
}

// Start consumers
async function main() {
  const emailConsumer = new EventConsumer('email_notifications', emailNotificationHandler)
  const analyticsConsumer = new EventConsumer('analytics_events', analyticsHandler)
  
  try {
    await emailConsumer.start()
    await analyticsConsumer.start()
    
    console.log('All consumers started successfully')
    
    // Graceful shutdown
    process.on('SIGINT', async () => {
      console.log('Shutting down consumers...')
      await emailConsumer.stop()
      await analyticsConsumer.stop()
      process.exit(0)
    })
    
  } catch (error) {
    console.error('Error starting consumers:', error)
  }
}

main()

class BatchConsumer {
  constructor(queueName, handler, batchSize = 10, batchTimeout = 5000) {
    this.queueName = queueName
    this.handler = handler
    this.batchSize = batchSize
    this.batchTimeout = batchTimeout
    this.connection = null
    this.channel = null
    this.messageBuffer = []
    this.batchTimer = null
  }
  
  async start() {
    try {
      this.connection = await createConnection()
      this.channel = await this.connection.createChannel()
      
      // Set prefetch to batch size
      await this.channel.prefetch(this.batchSize)
      
      await this.channel.assertExchange('events', 'topic', { durable: true })
      await this.channel.assertQueue(this.queueName, { durable: true })
      await this.channel.bindQueue(this.queueName, 'events', 'batch.*')
      
      await this.channel.consume(this.queueName, this.handleMessage.bind(this), {
        noAck: false
      })
      
      console.log(`Batch consumer started: ${this.queueName}`)
      
    } catch (error) {
      console.error('Failed to start batch consumer:', error)
      throw error
    }
  }
  
  handleMessage(message) {
    try {
      const event = JSON.parse(message.content.toString())
      this.messageBuffer.push({ event, message })
      
      // If buffer is full, process immediately
      if (this.messageBuffer.length >= this.batchSize) {
        await this.processBatch()
      } else {
        // Set timer to process batch if timeout
        this.scheduleBatchProcessing()
      }
      
    } catch (error) {
      console.error('Error parsing message:', error)
      this.channel.nack(message, false, false) // Don't requeue malformed messages
    }
  }
  
  scheduleBatchProcessing() {
    if (this.batchTimer) {
      clearTimeout(this.batchTimer)
    }
    
    this.batchTimer = setTimeout(async () => {
      if (this.messageBuffer.length > 0) {
        await this.processBatch()
      }
    }, this.batchTimeout)
  }
  
  async processBatch() {
    if (this.batchTimer) {
      clearTimeout(this.batchTimer)
      this.batchTimer = null
    }
    
    if (this.messageBuffer.length === 0) return
    
    const batch = this.messageBuffer.splice(0)
    
    try {
      console.log(`Processing batch of ${batch.length} messages`)
      
      // Process all messages in batch
      await this.handler(batch.map(item => item.event))
      
      // Acknowledge all messages
      batch.forEach(item => {
        this.channel.ack(item.message)
      })
      
      console.log(`Batch processed successfully`)
      
    } catch (error) {
      console.error('Error processing batch:', error)
      
      // Negative acknowledgment for all messages
      batch.forEach(item => {
        this.channel.nack(item.message, false, true)
      })
    }
  }
  
  async stop() {
    if (this.batchTimer) {
      clearTimeout(this.batchTimer)
    }
    
    // Process remaining messages
    await this.processBatch()
    
    if (this.channel) {
      await this.channel.cancel(this.queueName)
      await this.channel.close()
    }
    
    if (this.connection) {
      await this.connection.close()
    }
  }
}

// Usage example: Database batch inserter
async function batchDatabaseInserter(events) {
  if (events.length === 0) return
  
  try {
    // Batch insert into database
    await db.events.insertMany(events.map(event => ({
      type: event.type,
      data: JSON.stringify(event.data),
      timestamp: new Date(event.timestamp),
      userId: event.data.userId,
      source: event.source
    })))
    
    console.log(`Inserted ${events.length} events into database`)
  } catch (error) {
    console.error('Database batch insert failed:', error)
    throw error
  }
}

// Start batch consumer
const batchConsumer = new BatchConsumer('database_batch', batchDatabaseInserter, 50, 2000)
batchConsumer.start().catch(console.error)

class WorkQueue {
  constructor(queueName, workerFunction, numWorkers = 3) {
    this.queueName = queueName
    this.workerFunction = workerFunction
    this.workers = []
    this.numWorkers = numWorkers
  }
  
  async start() {
    for (let i = 0; i < this.numWorkers; i++) {
      const worker = new Worker(this.queueName, this.workerFunction, i)
      await worker.start()
      this.workers.push(worker)
    }
    
    console.log(`Started ${this.numWorkers} workers for queue: ${this.queueName}`)
  }
  
  async stop() {
    await Promise.all(this.workers.map(worker => worker.stop()))
  }
}

class Worker {
  constructor(queueName, workerFunction, workerId) {
    this.queueName = queueName
    this.workerFunction = workerFunction
    this.workerId = workerId
    this.connection = null
    this.channel = null
    this.isProcessing = false
  }
  
  async start() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    // Set prefetch to 1 for fair dispatching
    await this.channel.prefetch(1)
    
    await this.channel.assertQueue(this.queueName, { durable: true })
    
    await this.channel.consume(this.queueName, async (msg) => {
      if (this.isProcessing) return
      
      this.isProcessing = true
      
      try {
        console.log(`Worker ${this.workerId} processing message`)
        
        await this.workerFunction(JSON.parse(msg.content.toString()))
        
        this.channel.ack(msg)
        console.log(`Worker ${this.workerId} completed message`)
        
      } catch (error) {
        console.error(`Worker ${this.workerId} error:`, error)
        this.channel.nack(msg, false, true) // Requeue
      } finally {
        this.isProcessing = false
      }
    }, {
      noAck: false
    })
    
    console.log(`Worker ${this.workerId} started`)
  }
  
  async stop() {
    if (this.channel) {
      await this.channel.cancel(this.queueName)
      await this.channel.close()
    }
    if (this.connection) {
      await this.connection.close()
    }
  }
}

// Usage
async function processTask(task) {
  console.log('Processing task:', task)
  
  // Simulate work
  await new Promise(resolve => setTimeout(resolve, Math.random() * 2000 + 1000))
  
  console.log('Task completed:', task.id)
}

const workQueue = new WorkQueue('task_queue', processTask, 5)
workQueue.start()

class TopicPublisher {
  constructor(exchangeName = 'app_events') {
    this.exchangeName = exchangeName
    this.connection = null
    this.channel = null
  }
  
  async connect() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    await this.channel.assertExchange(this.exchangeName, 'topic', {
      durable: true
    })
    
    await this.channel.confirmChannel()
  }
  
  async publish(topic, event, options = {}) {
    const routingKey = topic
    const message = {
      id: generateId(),
      timestamp: new Date().toISOString(),
      event,
      source: 'event-service'
    }
    
    return this.channel.publish(
      this.exchangeName,
      routingKey,
      Buffer.from(JSON.stringify(message)),
      {
        persistent: true,
        messageId: message.id,
        headers: {
          'content-type': 'application/json',
          'event-topic': topic,
          ...options.headers
        }
      }
    )
  }
}

class TopicSubscriber {
  constructor(exchangeName, pattern, handler) {
    this.exchangeName = exchangeName
    this.pattern = pattern
    this.handler = handler
    this.connection = null
    this.channel = null
    this.queueName = null
  }
  
  async start() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    await this.channel.assertExchange(this.exchangeName, 'topic', {
      durable: true
    })
    
    // Create unique queue for this subscriber
    this.queueName = `${this.pattern.replace('*', 'wildcard')}_${Date.now()}`
    
    await this.channel.assertQueue(this.queueName, {
      durable: true,
      exclusive: false,
      autoDelete: false
    })
    
    // Bind with pattern
    await this.channel.bindQueue(this.queueName, this.exchangeName, this.pattern)
    
    await this.channel.consume(this.queueName, async (msg) => {
      try {
        const event = JSON.parse(msg.content.toString())
        
        console.log(`Received event: ${msg.fields.routingKey}`, event)
        
        await this.handler(event, {
          routingKey: msg.fields.routingKey,
          exchange: msg.fields.exchange,
          timestamp: msg.properties.timestamp
        })
        
        this.channel.ack(msg)
        
      } catch (error) {
        console.error('Error processing event:', error)
        this.channel.nack(msg, false, false)
      }
    }, {
      noAck: false
    })
    
    console.log(`Topic subscriber started for pattern: ${this.pattern}`)
  }
}

// Usage example
async function userEventHandler(event, context) {
  switch (context.routingKey) {
    case 'user.created':
      console.log('New user created:', event.event.data)
      break
    case 'user.updated':
      console.log('User updated:', event.event.data)
      break
    case 'user.deleted':
      console.log('User deleted:', event.event.data)
      break
  }
}

const publisher = new TopicPublisher()
await publisher.connect()

// Create subscribers for different patterns
const userSubscriber = new TopicSubscriber('app_events', 'user.*', userEventHandler)
const orderSubscriber = new TopicSubscriber('app_events', 'order.*', orderEventHandler)

await userSubscriber.start()
await orderSubscriber.start()

// Publish events
await publisher.publish('user.created', { userId: '123', email: 'user@example.com' })
await publisher.publish('order.created', { orderId: '456', amount: 99.99 })

class DLQManager {
  constructor() {
    this.connection = null
    this.channel = null
  }
  
  async setup() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    // Declare main exchange
    await this.channel.assertExchange('events', 'topic', { durable: true })
    
    // Declare dead letter exchange
    await this.channel.assertExchange('events_dlx', 'topic', { durable: true })
    
    // Declare dead letter queue
    await this.channel.assertQueue('events_dlq', {
      durable: true,
      arguments: {
        'x-message-ttl': 604800000, // 7 days
        'x-dead-letter-exchange': 'events',
        'x-dead-letter-routing-key': 'retry'
      }
    })
    
    // Bind DLQ to DLX
    await this.channel.bindQueue('events_dlq', 'events_dlx', '#')
    
    // Declare processing queue with DLQ
    await this.channel.assertQueue('events_processing', {
      durable: true,
      arguments: {
        'x-dead-letter-exchange': 'events_dlx',
        'x-dead-letter-routing-key': 'events_processing'
      }
    })
    
    // Bind processing queue to main exchange
    await this.channel.bindQueue('events_processing', 'events', 'important.*')
    
    // Set up consumer for processing queue
    await this.channel.consume('events_processing', async (msg) => {
      try {
        const event = JSON.parse(msg.content.toString())
        
        console.log('Processing important event:', event)
        
        // Simulate processing that might fail
        if (Math.random() < 0.2) { // 20% failure rate
          throw new Error('Simulated processing failure')
        }
        
        await this.processEvent(event)
        this.channel.ack(msg)
        
      } catch (error) {
        console.error('Processing failed:', error)
        this.channel.nack(msg, false, false) // Send to DLQ
      }
    }, {
      noAck: false
    })
    
    // Set up consumer for DLQ
    await this.channel.consume('events_dlq', async (msg) => {
      const failedEvent = JSON.parse(msg.content.toString())
      
      console.log('Failed event in DLQ:', failedEvent)
      
      // Extract failure information
      const failureInfo = {
        originalEvent: failedEvent,
        failureReason: msg.properties.headers['x-first-death-reason'],
        failedAt: new Date(parseInt(msg.properties.headers['x-first-death-'])),
        retryCount: parseInt(msg.properties.headers['x-death'].filter(x => x.count)[0]?.count || 0)
      }
      
      await this.handleFailedEvent(failureInfo)
      this.channel.ack(msg)
    }, {
      noAck: false
    })
    
    console.log('DLQ setup complete')
  }
  
  async processEvent(event) {
    // Your business logic here
    console.log('Successfully processed event:', event.id)
  }
  
  async handleFailedEvent(failureInfo) {
    console.log('Handling failed event:', failureInfo)
    
    // Store in database for manual review
    await db.failedEvents.create({
      eventId: failureInfo.originalEvent.id,
      reason: failureInfo.failureReason,
      failedAt: failureInfo.failedAt,
      retryCount: failureInfo.retryCount,
      eventData: failureInfo.originalEvent
    })
    
    // Send alert
    await sendAlert(`Event processing failed: ${failureInfo.originalEvent.id}`)
  }
  
  async sendAlert(message) {
    // Alert implementation (email, Slack, etc.)
    console.log('ALERT:', message)
  }
}

// Start DLQ manager
const dlqManager = new DLQManager()
dlqManager.setup().catch(console.error)

class PriorityPublisher {
  constructor() {
    this.connection = null
    this.channel = null
  }
  
  async connect() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    // Declare exchange with priority support
    await this.channel.assertExchange('priority_events', 'direct', {
      durable: true
    })
  }
  
  async publish(routingKey, message, priority = 5) {
    const messageData = {
      id: generateId(),
      timestamp: new Date().toISOString(),
      data: message,
      priority
    }
    
    return this.channel.publish(
      'priority_events',
      routingKey,
      Buffer.from(JSON.stringify(messageData)),
      {
        priority, // 0-9, higher = more important
        persistent: true,
        messageId: messageData.id,
        headers: {
          'content-type': 'application/json',
          'priority': priority.toString()
        }
      }
    )
  }
}

class PriorityConsumer {
  constructor(queueName, handler) {
    this.queueName = queueName
    this.handler = handler
    this.connection = null
    this.channel = null
  }
  
  async start() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    // Declare queue with max priority
    await this.channel.assertQueue(this.queueName, {
      durable: true,
      arguments: {
        'x-max-priority': 10 // Maximum priority value
      }
    })
    
    // Bind to priority exchange
    await this.channel.bindQueue(this.queueName, 'priority_events', this.queueName)
    
    await this.channel.consume(this.queueName, async (msg) => {
      try {
        const message = JSON.parse(msg.content.toString())
        const priority = msg.properties.priority || 5
        
        console.log(`Processing message with priority ${priority}:`, message)
        
        await this.handler(message, priority)
        this.channel.ack(msg)
        
      } catch (error) {
        console.error('Error processing priority message:', error)
        this.channel.nack(msg, false, false)
      }
    }, {
      noAck: false
    })
    
    console.log(`Priority consumer started: ${this.queueName}`)
  }
}

// Usage example
async function priorityHandler(message, priority) {
  // Handle based on priority
  if (priority >= 8) {
    console.log('HIGH PRIORITY - Immediate processing')
    await processImmediately(message)
  } else if (priority >= 5) {
    console.log('MEDIUM PRIORITY - Normal processing')
    await processNormally(message)
  } else {
    console.log('LOW PRIORITY - Background processing')
    await processInBackground(message)
  }
}

const publisher = new PriorityPublisher()
const consumer = new PriorityConsumer('priority_queue', priorityHandler)

await publisher.connect()
await consumer.start()

// Publish messages with different priorities
await publisher.publish('priority_queue', { task: 'urgent_task' }, 9)
await publisher.publish('priority_queue', { task: 'normal_task' }, 5)
await publisher.publish('priority_queue', { task: 'background_task' }, 1)

class ResilientConnection {
  constructor(config) {
    this.config = config
    this.connection = null
    this.channels = new Map()
    this.reconnectTimer = null
    this.isReconnecting = false
    this.connectionPromise = null
  }
  
  async getConnection() {
    if (this.connection && this.connection.connection.serverProperties) {
      return this.connection
    }
    
    if (this.connectionPromise) {
      return this.connectionPromise
    }
    
    this.connectionPromise = this.establishConnection()
    return this.connectionPromise
  }
  
  async establishConnection() {
    while (true) {
      try {
        const connection = await amqp.connect(this.config.url, {
          timeout: this.config.timeout || 30000,
          heartbeat: this.config.heartbeat || 60,
          clientProperties: {
            product_name: 'resilient-app',
            connection_name: 'resilient-connection'
          }
        })
        
        this.setupConnectionHandlers(connection)
        this.connection = connection
        this.connectionPromise = null
        
        console.log('Connection established successfully')
        return connection
        
      } catch (error) {
        console.error('Connection failed:', error)
        
        // Exponential backoff
        const delay = Math.min(1000 * Math.pow(2, this.getRetryCount()), 30000)
        console.log(`Retrying connection in ${delay}ms...`)
        
        await new Promise(resolve => setTimeout(resolve, delay))
      }
    }
  }
  
  setupConnectionHandlers(connection) {
    connection.on('error', (error) => {
      console.error('Connection error:', error)
      this.handleConnectionError(error)
    })
    
    connection.on('close', (reason) => {
      console.log('Connection closed:', reason)
      this.handleConnectionClose(reason)
    })
    
    connection.on('blocked', (reason) => {
      console.warn('Connection blocked:', reason)
    })
    
    connection.on('unblocked', () => {
      console.log('Connection unblocked')
    })
  }
  
  handleConnectionError(error) {
    if (this.isReconnecting) return
    
    console.error('Handling connection error:', error)
    this.scheduleReconnect()
  }
  
  handleConnectionClose(reason) {
    if (this.isReconnecting) return
    
    console.log('Handling connection close:', reason)
    this.connection = null
    this.channels.clear()
    this.scheduleReconnect()
  }
  
  scheduleReconnect() {
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer)
    }
    
    this.isReconnecting = true
    
    this.reconnectTimer = setTimeout(async () => {
      console.log('Attempting to reconnect...')
      this.isReconnecting = false
      
      try {
        await this.establishConnection()
      } catch (error) {
        console.error('Reconnection failed:', error)
        this.scheduleReconnect()
      }
    }, 5000) // Wait 5 seconds before reconnecting
  }
  
  async getChannel() {
    const connection = await this.getConnection()
    
    // Check if we already have a channel for this connection
    const existingChannel = this.channels.get(connection)
    if (existingChannel) {
      return existingChannel
    }
    
    // Create new channel
    const channel = await connection.createChannel()
    this.channels.set(connection, channel)
    
    // Set up channel error handler
    channel.on('error', (error) => {
      console.error('Channel error:', error)
    })
    
    channel.on('close', () => {
      console.log('Channel closed')
      this.channels.delete(connection)
    })
    
    return channel
  }
  
  getRetryCount() {
    // Implement retry count tracking
    return this.retryCount = (this.retryCount || 0) + 1
  }
}

class ReliableConsumer {
  constructor(queueName, handler, options = {}) {
    this.queueName = queueName
    this.handler = handler
    this.options = {
      maxRetries: options.maxRetries || 3,
      retryDelay: options.retryDelay || 5000,
      visibilityTimeout: options.visibilityTimeout || 30000,
      deadLetterQueue: `${queueName}_dlq`,
      ...options
    }
    
    this.connection = null
    this.channel = null
    this.processingMessages = new Map()
  }
  
  async start() {
    this.connection = await createConnection()
    this.channel = await this.connection.createChannel()
    
    // Declare dead letter exchange and queue
    await this.channel.assertExchange(`${this.queueName}_dlx`, 'direct', {
      durable: true
    })
    
    await this.channel.assertQueue(this.options.deadLetterQueue, {
      durable: true,
      arguments: {
        'x-message-ttl': 604800000 // 7 days
      }
    })
    
    await this.channel.bindQueue(
      this.options.deadLetterQueue,
      `${this.queueName}_dlx`,
      this.options.deadLetterQueue
    )
    
    // Declare main queue with DLQ
    await this.channel.assertQueue(this.queueName, {
      durable: true,
      arguments: {
        'x-dead-letter-exchange': `${this.queueName}_dlx`,
        'x-dead-letter-routing-key': this.options.deadLetterQueue,
        'x-message-ttl': this.options.visibilityTimeout
      }
    })
    
    await this.channel.consume(this.queueName, this.handleMessage.bind(this), {
      noAck: false
    })
    
    // Set up visibility timeout checker
    setInterval(() => this.checkTimeouts(), 5000)
  }
  
  async handleMessage(message) {
    const messageId = message.properties.messageId || message.fields.deliveryTag
    const startTime = Date.now()
    
    // Track message processing
    this.processingMessages.set(messageId, {
      message,
      startTime,
      retries: this.getMessageRetries(message)
    })
    
    try {
      console.log(`Processing message: ${messageId}`)
      
      // Process the message
      await this.handler(JSON.parse(message.content.toString()), message)
      
      // Remove from tracking and acknowledge
      this.processingMessages.delete(messageId)
      this.channel.ack(message)
      
      console.log(`Message processed successfully: ${messageId}`)
      
    } catch (error) {
      console.error(`Error processing message ${messageId}:`, error)
      
      const processingInfo = this.processingMessages.get(messageId)
      
      if (processingInfo.retries < this.options.maxRetries) {
        // Retry the message
        console.log(`Retrying message ${messageId} (attempt ${processingInfo.retries + 1})`)
        
        // Update retry count in headers
        this.processingMessages.set(messageId, {
          ...processingInfo,
          retries: processingInfo.retries + 1
        })
        
        // Requeue with delay
        setTimeout(() => {
          this.channel.nack(message, false, true)
        }, this.options.retryDelay)
        
      } else {
        // Max retries reached - send to DLQ
        console.log(`Max retries reached for message ${messageId}, sending to DLQ`)
        
        this.processingMessages.delete(messageId)
        this.channel.nack(message, false, false) // Don't requeue
        
        // Log to error tracking system
        await this.logProcessingError(messageId, error, processingInfo.retries)
      }
    }
  }
  
  getMessageRetries(message) {
    const retryHeader = message.properties.headers['x-retry-count']
    return retryHeader ? parseInt(retryHeader) : 0
  }
  
  checkTimeouts() {
    const now = Date.now()
    
    for (const [messageId, processingInfo] of this.processingMessages) {
      if (now - processingInfo.startTime > this.options.visibilityTimeout) {
        console.warn(`Message ${messageId} processing timeout`)
        
        // Treat as failed and send to DLQ
        this.channel.nack(processingInfo.message, false, false)
        this.processingMessages.delete(messageId)
        
        // Log timeout
        this.logTimeout(messageId, processingInfo)
      }
    }
  }
  
  async logProcessingError(messageId, error, retries) {
    // Log to your monitoring system
    await errorLogger.log({
      messageId,
      error: error.message,
      stack: error.stack,
      retries,
      timestamp: new Date().toISOString(),
      queue: this.queueName
    })
  }
  
  async logTimeout(messageId, processingInfo) {
    // Log timeout to monitoring system
    await errorLogger.log({
      messageId,
      error: 'Processing timeout',
      processingTime: Date.now() - processingInfo.startTime,
      retries: processingInfo.retries,
      timestamp: new Date().toISOString(),
      queue: this.queueName
    })
  }
}