RabbitMQ #

gem install bunny

# For Sneakers (a worker framework for Rails)
gem install sneakers

# Gemfile
gem 'bunny',    '~> 2.22'   # AMQP client
gem 'sneakers', '~> 2.12'   # worker framework (optional, for Rails)

require 'bunny'

# Basic connection
connection = Bunny.new(
  host:     "localhost",
  port:     5672,
  vhost:    "/",
  username: "guest",
  password: "guest"
)

# Connection via URL (more concise)
connection = Bunny.new(
  ENV.fetch("RABBITMQ_URL", "amqp://guest:***@localhost:5672")
)

# Connection with TLS (production)
connection = Bunny.new(
  host:     "rabbitmq.example.com",
  port:     5671,                    # TLS port
  username: ENV["RABBITMQ_USER"],
  password: ENV["RABBITMQ_PASSWORD"],
  tls:      true,
  tls_ca_certificates: ["/path/to/ca-cert.pem"]
)

# Additional options
connection = Bunny.new(
  "amqp://localhost",
  heartbeat:              10,     # heartbeat every 10 seconds
  connection_timeout:     5,      # connection timeout
  read_timeout:           30,     # read timeout
  automatically_recover:  true,   # auto-reconnect if disconnected
  recovery_attempts:      10,     # max 10 reconnect attempts
  recovery_delay:         2       # 2 second delay between attempts
)

connection.start

# Channel — the AMQP communication unit, one connection can have many channels
# One channel per thread is the recommended practice
channel = connection.create_channel

puts "Connected to RabbitMQ #{connection.server_properties['version']}"

# Close the connection
connection.close

channel = connection.create_channel

# Declare an exchange
exchange = channel.direct(
  "orders_exchange",
  durable:     true,    # survives RabbitMQ restarts
  auto_delete: false
)

# Declare a queue and binding
new_queue = channel.queue(
  "new_orders",
  durable:    true,
  arguments: { "x-message-ttl" => 86_400_000 }  # 24 hour TTL
)

# Bind the queue to the exchange with a routing key
new_queue.bind(exchange, routing_key: "order.new")

# Send a message to the exchange with a routing key
exchange.publish(
  JSON.generate({ order_id: 123, total: 150_000 }),
  routing_key:  "order.new",
  persistent:   true,    # the message is stored on disk (not lost on restart)
  content_type: "application/json"
)

# Fanout — send to all bound queues, ignoring routing keys
fanout = channel.fanout(
  "notification_broadcast",
  durable: true
)

# Create several queues and bind them to the fanout
["email_notifications", "sms_notifications", "push_notifications"].each do |queue_name|
  q = channel.queue(queue_name, durable: true)
  q.bind(fanout)   # no routing key needed for fanout
end

# One publish → all queues receive it
fanout.publish(
  JSON.generate({ event: "flash_sale", discount: 50, ends_at: "2024-08-15T23:59:59" }),
  persistent: true
)

# Topic exchange — routing keys with patterns
# * = exactly one word
# # = zero or more words
topic = channel.topic("event_bus", durable: true)

# Bindings with patterns
channel.queue("all_orders", durable: true)
  .bind(topic, routing_key: "order.#")      # all order events

channel.queue("new_and_paid_orders", durable: true)
  .bind(topic, routing_key: "order.new")
  .bind(topic, routing_key: "order.paid")

channel.queue("all_email_notifications", durable: true)
  .bind(topic, routing_key: "*.email.#")      # email from any domain

# Publish with hierarchical routing keys
topic.publish(
  JSON.generate({ order_id: 456 }),
  routing_key: "order.new",
  persistent:  true
)

topic.publish(
  JSON.generate({ user_id: 99, template: "welcome" }),
  routing_key: "user.email.welcome",
  persistent:  true
)

require 'bunny'
require 'json'

class OrderProducer
  def initialize
    @connection = Bunny.new(ENV["RABBITMQ_URL"])
    @connection.start
    @channel  = @connection.create_channel

    # Enable publisher confirms — confirmation from RabbitMQ that the message was received
    @channel.confirm_select

    @exchange = @channel.topic("event_bus", durable: true)
  end

  def send_order_created(order)
    payload = JSON.generate({
      event:      "order_created",
      order_id:   order.id,
      user_id:    order.user_id,
      total:      order.total,
      items:      order.items.map { |i| { product_id: i.product_id, quantity: i.quantity } },
      timestamp:  Time.now.iso8601
    })

    @exchange.publish(
      payload,
      routing_key:  "order.new",
      persistent:   true,
      content_type: "application/json",
      message_id:   SecureRandom.uuid,   # unique ID for idempotency
      timestamp:    Time.now.to_i
    )

    # Wait for confirmation from RabbitMQ (publisher confirms)
    unless @channel.wait_for_confirms
      raise "RabbitMQ did not confirm the message delivery!"
    end

    true
  rescue Bunny::Exception => e
    Rails.logger.error "Failed to send message to RabbitMQ: #{e.message}"
    false
  end

  def close
    @connection.close
  end
end

require 'bunny'
require 'json'

connection = Bunny.new(ENV["RABBITMQ_URL"])
connection.start

channel = connection.create_channel

# Prefetch count — how many messages are sent to a consumer before an ack
# A value of 1 = the consumer only receives 1 message, processes it, acks, then gets the next
# Important for fair load distribution between consumers
channel.prefetch(1)

queue = channel.queue("new_orders", durable: true)

puts "Waiting for messages in queue '#{queue.name}'..."

# Subscribe — the consumer loop (blocking)
queue.subscribe(manual_ack: true, block: true) do |delivery_info, properties, body|
  begin
    data = JSON.parse(body, symbolize_names: true)
    puts "Processing order #{data[:order_id]}"

    # Process the message
    process_order(data)

    # ACK — tell RabbitMQ the message was processed successfully, remove it from the queue
    channel.ack(delivery_info.delivery_tag)
    puts "ACK: order #{data[:order_id]} done"

  rescue JSON::ParserError => e
    puts "Invalid message: #{e.message}"
    # NACK without requeue — send to a Dead Letter Exchange if configured
    channel.nack(delivery_info.delivery_tag, false, false)

  rescue => e
    puts "Error: #{e.message}"
    # NACK with requeue — return to the queue to try again
    channel.nack(delivery_info.delivery_tag, false, true)
  end
end

require 'bunny'

connection = Bunny.new(ENV["RABBITMQ_URL"])
connection.start

# Create N worker threads, each with its own channel
WORKER_COUNT = 5

workers = WORKER_COUNT.times.map do |i|
  Thread.new do
    ch = connection.create_channel
    ch.prefetch(1)
    q  = ch.queue("new_orders", durable: true)

    q.subscribe(manual_ack: true) do |delivery_info, _properties, body|
      begin
        data = JSON.parse(body, symbolize_names: true)
        puts "[Worker #{i}] Processing order #{data[:order_id]}"

        process_order(data)
        ch.ack(delivery_info.delivery_tag)

      rescue => e
        puts "[Worker #{i}] Error: #{e.message}"
        ch.nack(delivery_info.delivery_tag, false, false)
      end
    end

    # The thread keeps running until a stop signal
    loop { sleep 1 }
  end
end

# Wait for all workers
trap("SIGTERM") { connection.close; exit }
trap("SIGINT")  { connection.close; exit }
workers.each(&:join)

# Three options after processing a message:

# 1. ACK — processed successfully, remove from the queue
channel.ack(delivery_info.delivery_tag)

# 2. NACK with requeue: true — return to the queue (another consumer can pick it up)
channel.nack(delivery_info.delivery_tag, false, true)
# CAREFUL: can cause an infinite loop if a message always fails!

# 3. NACK with requeue: false — remove from the queue
#    If a Dead Letter Exchange exists → goes to the DLX
#    If not → the message is lost
channel.nack(delivery_info.delivery_tag, false, false)

# 4. Reject — same as NACK but only for a single message
channel.reject(delivery_info.delivery_tag, false)  # false = no requeue

# Recommended strategy:
# - Success → ACK
# - Temporary error (network, timeout) → NACK requeue: true (with backoff)
# - Permanent error (invalid data) → NACK requeue: false → goes to the DLX

channel = connection.create_channel

# 1. Create a Dead Letter Exchange and Queue
dlx = channel.direct("orders_dlx", durable: true)
dlq = channel.queue("orders_dead_letter", durable: true)
dlq.bind(dlx, routing_key: "new_orders")

# 2. Create the main queue with DLX configuration
main_queue = channel.queue(
  "new_orders",
  durable:   true,
  arguments: {
    "x-dead-letter-exchange"    => "orders_dlx",   # DLX name
    "x-dead-letter-routing-key" => "new_orders",   # routing key on the DLX
    "x-message-ttl"             => 3_600_000,      # 1 hour TTL (optional)
    "x-max-retries"             => 3               # custom header for tracking
  }
)

# 3. Bind the main queue to the main exchange
exchange = channel.topic("event_bus", durable: true)
main_queue.bind(exchange, routing_key: "order.new")

# 4. Consumer with retry logic and DLX
main_queue.subscribe(manual_ack: true, block: true) do |delivery_info, properties, body|
  headers       = properties.headers || {}
  death_count   = headers["x-death"]&.first&.dig("count") || 0

  begin
    process_order(JSON.parse(body, symbolize_names: true))
    channel.ack(delivery_info.delivery_tag)

  rescue => e
    puts "Error (attempt #{death_count + 1}): #{e.message}"

    if death_count >= 2   # failed 3 times (0, 1, 2)
      puts "Giving up after 3 attempts, sending to DLX"
      channel.nack(delivery_info.delivery_tag, false, false)  # → DLX
    else
      # Return to the queue with a delay (simple implementation)
      sleep(2 ** death_count)   # 1s, 2s, 4s
      channel.nack(delivery_info.delivery_tag, false, true)   # requeue
    end
  end
end

# 5. DLQ consumer — monitor and process truly failed messages
dlq.subscribe(manual_ack: true) do |delivery_info, properties, body|
  puts "Message in DLQ: #{body}"

  # Log to a monitoring system
  Monitoring.alert("Message in DLQ", body: body, headers: properties.headers)

  # Optional: notify developers or save to a database for analysis
  FailedOrder.create!(payload: body, failed_at: Time.now)

  channel.ack(delivery_info.delivery_tag)
end

# Create a priority queue with a maximum of 10 priority levels
priority_queue = channel.queue(
  "priority_notifications",
  durable:   true,
  arguments: { "x-max-priority" => 10 }
)

exchange = channel.direct("notification_exchange", durable: true)
priority_queue.bind(exchange, routing_key: "notifications")

# Send messages with different priorities
# Priority 0 (lowest) through 10 (highest)
exchange.publish(
  JSON.generate({ message: "Regular info", type: "info" }),
  routing_key: "notifications",
  priority:    1,
  persistent:  true
)

exchange.publish(
  JSON.generate({ message: "Server down!", type: "critical" }),
  routing_key: "notifications",
  priority:    10,   # highest — processed first
  persistent:  true
)

exchange.publish(
  JSON.generate({ message: "Stock running low", type: "warning" }),
  routing_key: "notifications",
  priority:    5,
  persistent:  true
)

# CLIENT — send a request and wait for the reply
class CalculatorRPCClient
  def initialize
    @connection = Bunny.new(ENV["RABBITMQ_URL"])
    @connection.start
    @channel  = @connection.create_channel

    # Exclusive queue for receiving replies (auto-deleted when disconnected)
    @reply_queue = @channel.queue("", exclusive: true)
    @exchange    = @channel.default_exchange

    # Hash to store waiting responses
    @lock    = Mutex.new
    @cond    = ConditionVariable.new
    @responses = {}

    # Subscribe to the reply queue
    @reply_queue.subscribe do |_di, properties, body|
      correlation_id = properties.correlation_id
      @lock.synchronize do
        @responses[correlation_id] = body
        @cond.signal
      end
    end
  end

  def factorial(n)
    correlation_id = SecureRandom.uuid

    @exchange.publish(
      n.to_s,
      routing_key:    "rpc_factorial",
      reply_to:       @reply_queue.name,
      correlation_id: correlation_id
    )

    # Wait for the response with a 5 second timeout
    @lock.synchronize do
      @cond.wait(@lock, 5)
      result = @responses.delete(correlation_id)
      raise "RPC timeout!" unless result
      JSON.parse(result)
    end
  end

  def close
    @connection.close
  end
end

# SERVER — receive requests, process them, send replies
def run_rpc_server
  connection = Bunny.new(ENV["RABBITMQ_URL"])
  connection.start
  channel = connection.create_channel
  channel.prefetch(1)

  queue = channel.queue("rpc_factorial", durable: false)

  queue.subscribe(manual_ack: true, block: true) do |di, properties, body|
    n     = body.to_i
    result = recursive_factorial(n)

    # Send the reply to the reply_to queue with the same correlation_id
    channel.default_exchange.publish(
      JSON.generate({ result: result, input: n }),
      routing_key:    properties.reply_to,
      correlation_id: properties.correlation_id
    )

    channel.ack(di.delivery_tag)
  end
end

# RPC usage
client = CalculatorRPCClient.new
puts client.factorial(10)   # => {"result"=>3628800, "input"=>10}
client.close

# Gemfile
# gem 'sneakers', '~> 2.12'

# config/initializers/sneakers.rb
Sneakers.configure(
  amqp:          ENV.fetch("RABBITMQ_URL", "amqp://localhost"),
  vhost:         "/",
  heartbeat:     10,
  prefetch:      10,
  workers:       4,
  threads:       4,
  log:           Rails.root.join("log", "sneakers.log"),
  pid_path:      Rails.root.join("tmp", "pids", "sneakers.pid"),
  daemonize:     false,
  metrics:       Sneakers::Metrics::LoggingMetrics.new
)

# app/workers/order_worker.rb
class OrderWorker
  include Sneakers::Worker

  from_queue "new_orders",
    durable:       true,
    ack:           :manual,    # manual acknowledgment
    timeout_job_after: 60,     # 60 second timeout per job
    exchange:      "event_bus",
    exchange_type: :topic,
    routing_key:   "order.new"

  def work(message)
    data = JSON.parse(message, symbolize_names: true)
    Rails.logger.info "Processing order: #{data[:order_id]}"

    ActiveRecord::Base.transaction do
      order = Order.find(data[:order_id])
      order.update!(status: :processed)
      SendOrderEmail.call(order)
      UpdateStock.call(order)
    end

    ack!   # success → ACK

  rescue ActiveRecord::RecordNotFound => e
    Rails.logger.error "Order not found: #{e.message}"
    reject!   # invalid data → don't requeue, send to the DLX

  rescue => e
    Rails.logger.error "Error: #{e.message}"
    requeue!  # temporary error → return to the queue
  end
end

# app/workers/notification_worker.rb
class NotificationWorker
  include Sneakers::Worker

  from_queue "email_notifications",
    durable:    true,
    ack:        :manual

  def work(message)
    data = JSON.parse(message, symbolize_names: true)
    NotificationMailer.send(data[:email], data[:subject], data[:body]).deliver_now
    ack!
  rescue => e
    Rails.logger.error "Failed to send notification: #{e.message}"
    reject!
  end
end

# Run all registered workers
bundle exec rake sneakers:run

# Or run specific workers
WORKERS=OrderWorker,NotificationWorker bundle exec rake sneakers:run

# As a daemon
bundle exec rake sneakers:run DAEMONIZE=true

# Handle SIGTERM for a clean shutdown
require 'bunny'

connection = Bunny.new(ENV["RABBITMQ_URL"])
connection.start
channel = connection.create_channel
channel.prefetch(1)

queue = channel.queue("new_orders", durable: true)
running = true
messages_in_flight = 0

trap("SIGTERM") do
  puts "SIGTERM received, waiting for in-flight messages to finish..."
  running = false
end

trap("SIGINT") do
  running = false
end

queue.subscribe(manual_ack: true) do |delivery_info, properties, body|
  break unless running

  messages_in_flight += 1
  begin
    process_order(JSON.parse(body))
    channel.ack(delivery_info.delivery_tag)
  rescue => e
    channel.nack(delivery_info.delivery_tag, false, true)
  ensure
    messages_in_flight -= 1
  end
end

# Wait for all messages to finish before closing
sleep(0.1) while messages_in_flight > 0
connection.close
puts "Shutdown complete"

require 'net/http'
require 'json'

class RabbitMQMonitor
  BASE_URL = "http://localhost:15672/api"

  def initialize(user: "guest", password: "guest")
    @user     = user
    @password = password
  end

  def queue_stats(queue_name, vhost: "%2F")
    response = get("/queues/#{vhost}/#{queue_name}")
    {
      messages_waiting: response["messages"],
      messages_unacked: response["messages_unacknowledged"],
      active_consumers: response["consumers"],
      publish_rate:     response.dig("message_stats", "publish_details", "rate")&.round(2)
    }
  end

  def all_queues(vhost: "%2F")
    get("/queues/#{vhost}").map do |q|
      { name: q["name"], messages: q["messages"], consumers: q["consumers"] }
    end
  end

  def health_check
    get("/healthchecks/node")["status"] == "ok"
  end

  private

  def get(path)
    uri  = URI("#{BASE_URL}#{path}")
    req  = Net::HTTP::Get.new(uri)
    req.basic_auth(@user, @password)
    resp = Net::HTTP.start(uri.host, uri.port) { |http| http.request(req) }
    JSON.parse(resp.body)
  end
end

monitor = RabbitMQMonitor.new
stats = monitor.queue_stats("new_orders")
puts "Messages waiting: #{stats[:messages_waiting]}"
puts "Active consumers: #{stats[:active_consumers]}"

if stats[:messages_waiting] > 1000
  Monitoring.alert("Queue is full: new_orders has #{stats[:messages_waiting]} messages!")
end