# Step 1: Create the dead letter queue
channel.queue_declare(
    queue="failed-orders",
    durable=True
)

# Step 2: Create main queue with DLQ configured
channel.queue_declare(
    queue="order-processing",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "",       # Use default exchange for DLQ routing
        "x-dead-letter-routing-key": "failed-orders",  # DLQ queue name
        "x-message-ttl": 3600000,           # Messages expire after 1 hour
        "x-max-length": 10000               # Max 10,000 messages in queue
    }
)

# Step 3: Consumer that handles failures
def process_order(ch, method, properties, body):
    try:
        order = json.loads(body)
        if order.get("total", 0) <= 0:
            raise ValueError("Invalid order total")
        process_valid_order(order)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"Order processing failed: {e}")
        # Reject → goes to failed-orders DLQ
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

# Step 4: Monitor the DLQ separately
def handle_failed_order(ch, method, properties, body):
    order = json.loads(body)
    print(f"ALERT: Order {order.get('id')} failed — manual review needed")
    # Log to Slack, create ticket, notify support team
    log_to_monitoring_system(order, properties.headers)
    ch.basic_ack(delivery_tag=method.delivery_tag)

# Multiple workers compete for messages in the same queue
# Only ONE worker processes each message

# Producer
for task in tasks:
    channel.basic_publish(
        exchange="",
        routing_key="work_queue",
        body=json.dumps(task),
        properties=pika.BasicProperties(delivery_mode=2)
    )

# Worker 1, Worker 2, Worker 3 — all consume from same queue
# Load automatically balanced across workers
channel.basic_qos(prefetch_count=1)  # Fair distribution
channel.basic_consume(queue="work_queue", on_message_callback=process_task)

# One event → many different consumers each get their own copy
# Use fanout exchange

channel.exchange_declare(exchange="user.registered", exchange_type="fanout")

# Publisher: User service
channel.basic_publish(
    exchange="user.registered",
    routing_key="",
    body=json.dumps({"user_id": 123, "email": "user@example.com", "name": "Rahul"})
)

# Consumer 1: Email service — sends welcome email
# Consumer 2: Analytics service — tracks new user
# Consumer 3: CRM service — creates customer record
# All three receive the same message independently

# Synchronous-feeling communication over async infrastructure
import uuid

def call_service(request_data):
    correlation_id = str(uuid.uuid4())
    callback_queue = channel.queue_declare(queue="", exclusive=True).method.queue

    channel.basic_publish(
        exchange="",
        routing_key="service_request_queue",
        properties=pika.BasicProperties(
            reply_to=callback_queue,           # Where to send the response
            correlation_id=correlation_id      # Match response to request
        ),
        body=json.dumps(request_data)
    )

    # Wait for response
    for method, properties, body in channel.consume(callback_queue, auto_ack=True):
        if properties.correlation_id == correlation_id:
            return json.loads(body)

# Process message after a delay (e.g., send reminder 24 hours after signup)
channel.queue_declare(
    queue="delayed_notifications",
    arguments={
        "x-message-ttl": 86400000,              # 24 hours in milliseconds
        "x-dead-letter-exchange": "",
        "x-dead-letter-routing-key": "send_notifications"  # Process after delay
    }
)

# Publish to delayed queue — will appear in send_notifications queue after 24h
channel.basic_publish(
    exchange="",
    routing_key="delayed_notifications",
    body=json.dumps({"user_id": 123, "type": "onboarding_reminder"})
)

// Install: npm install amqplib
const amqp = require("amqplib");

// ─── Producer ─────────────────────────────────
async function sendOrderToQueue(order) {
    const connection = await amqp.connect(process.env.RABBITMQ_URL || "amqp://localhost");
    const channel = await connection.createChannel();

    const queueName = "order_processing";
    await channel.assertQueue(queueName, { durable: true });

    const message = JSON.stringify(order);
    channel.sendToQueue(queueName, Buffer.from(message), {
        persistent: true,           // Survive RabbitMQ restart
        contentType: "application/json",
        timestamp: Date.now()
    });

    console.log(`Order ${order.id} sent to queue`);

    setTimeout(() => {
        channel.close();
        connection.close();
    }, 500);
}

// Express route
app.post("/orders", async (req, res) => {
    const order = await saveOrderToDatabase(req.body);
    await sendOrderToQueue(order);
    res.status(201).json({ success: true, orderId: order.id });
    // Returns immediately — order processing is async!
});

// ─── Consumer ─────────────────────────────────
async function startOrderConsumer() {
    const connection = await amqp.connect(process.env.RABBITMQ_URL || "amqp://localhost");
    const channel = await connection.createChannel();

    await channel.assertQueue("order_processing", { durable: true });
    channel.prefetch(5);   // Process 5 orders simultaneously

    console.log("Order consumer started. Waiting for orders...");

    channel.consume("order_processing", async (msg) => {
        if (!msg) return;

        const order = JSON.parse(msg.content.toString());
        console.log(`Processing order: ${order.id}`);

        try {
            await Promise.all([
                sendConfirmationEmail(order),
                updateInventory(order),
                notifyWarehouse(order),
                generateInvoice(order)
            ]);

            channel.ack(msg);   // Success — remove from queue
            console.log(`Order ${order.id} processed successfully`);

        } catch (error) {
            console.error(`Order ${order.id} failed:`, error.message);

            const retryCount = (msg.properties.headers?.["x-retry-count"] || 0);
            if (retryCount < 3) {
                // Requeue with retry count
                channel.nack(msg, false, false);  // To DLQ
            } else {
                channel.nack(msg, false, false);  // Max retries exceeded → DLQ
            }
        }
    });

    // Handle connection errors
    connection.on("error", (err) => {
        console.error("RabbitMQ connection error:", err);
        setTimeout(startOrderConsumer, 5000);  // Reconnect after 5 seconds
    });
}

startOrderConsumer();

# Run with Docker (easiest setup)
docker run -d \
    --name rabbitmq \
    -p 5672:5672 \       # AMQP port
    -p 15672:15672 \     # Management UI port
    -e RABBITMQ_DEFAULT_USER=admin \
    -e RABBITMQ_DEFAULT_PASS=password \
    rabbitmq:3-management

# Access Management UI at: http://localhost:15672
# Login: admin / password

# Install Python client
pip install pika

# Install Node.js client
npm install amqplib

# compose.yml
services:
  app:
    build: .
    environment:
      - RABBITMQ_URL=amqp://admin:password@rabbitmq:5672
    depends_on:
      rabbitmq:
        condition: service_healthy

  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=password
    healthcheck:
      test: rabbitmq-diagnostics -q ping
      interval: 10s
      timeout: 5s
      retries: 5
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq

volumes:
  rabbitmq_data: