import amqp from 'amqplib';

interface OrderMessage {
  orderId: string;
  customerId: string;
  total: number;
  timestamp: Date;
}

class RabbitMQProducer {
  private connection: amqp.Connection | null = null;
  private channel: amqp.Channel | null = null;

  async connect(url: string = 'amqp://localhost'): Promise<void> {
    this.connection = await amqp.connect(url);
    this.channel = await this.connection.createChannel();

    // Declare exchange (idempotent)
    const exchangeName = 'orders';
    await this.channel.assertExchange(exchangeName, 'topic', { durable: true });
  }

  async publishOrder(message: OrderMessage): Promise<boolean> {
    if (!this.channel) {
      throw new Error('Producer not connected');
    }

    const exchangeName = 'orders';
    const routingKey = `orders.${message.customerId}.created`;
    const messageBuffer = Buffer.from(JSON.stringify(message));

    // Publish with confirmation
    return this.channel.publish(
      exchangeName,
      routingKey,
      messageBuffer,
      {
        persistent: true,
        contentType: 'application/json',
        timestamp: Date.now(),
        // Generate unique message ID for idempotency
        messageId: `${message.orderId}-${Date.now()}`,
      }
    );
  }

  async close(): Promise<void> {
    if (this.channel) {
      await this.channel.close();
    }
    if (this.connection) {
      await this.connection.close();
    }
  }
}

// Usage
const producer = new RabbitMQProducer();
await producer.connect('amqp://guest:guest@localhost');

await producer.publishOrder({
  orderId: 'ORD-12345',
  customerId: 'CUST-789',
  total: 149.99,
  timestamp: new Date(),
});

await producer.close();

// RabbitMQProducer.java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.AMQP;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeoutException;

public class RabbitMQProducer {
    private Connection connection;
    private Channel channel;
    private static final ObjectMapper mapper = new ObjectMapper();

    public record OrderMessage(String orderId, String customerId, double total, Date timestamp) {}

    public void connect(String url) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri(url);
        this.connection = factory.newConnection();
        this.channel = connection.createChannel();

        // Declare exchange (idempotent)
        channel.exchangeDeclare("orders", "topic", true);
    }

    public boolean publishOrder(OrderMessage message) throws IOException {
        if (channel == null) throw new IllegalStateException("Producer not connected");

        String exchangeName = "orders";
        String routingKey = "orders." + message.customerId() + ".created";
        byte[] body = mapper.writeValueAsBytes(message);

        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .deliveryMode(2) // persistent
            .contentType("application/json")
            .timestamp(new Date())
            .messageId(message.orderId() + "-" + System.currentTimeMillis())
            .build();

        channel.basicPublish(exchangeName, routingKey, props, body);
        return true;
    }

    public void close() throws IOException, TimeoutException {
        if (channel != null) channel.close();
        if (connection != null) connection.close();
    }

    // Usage
    public static void main(String[] args) throws Exception {
        RabbitMQProducer producer = new RabbitMQProducer();
        producer.connect("amqp://guest:guest@localhost");

        producer.publishOrder(new OrderMessage("ORD-12345", "CUST-789", 149.99, new Date()));

        producer.close();
    }
}

# rabbitmq_producer.py
import aio_pika
import json
from datetime import datetime
from dataclasses import dataclass, asdict


@dataclass
class OrderMessage:
    order_id: str
    customer_id: str
    total: float
    timestamp: str


class RabbitMQProducer:
    def __init__(self):
        self.connection = None
        self.channel = None

    async def connect(self, url: str = "amqp://localhost") -> None:
        self.connection = await aio_pika.connect_robust(url)
        self.channel = await self.connection.channel()

        # Declare exchange (idempotent)
        self.exchange = await self.channel.declare_exchange(
            "orders", aio_pika.ExchangeType.TOPIC, durable=True
        )

    async def publish_order(self, message: OrderMessage) -> None:
        if not self.channel:
            raise RuntimeError("Producer not connected")

        routing_key = f"orders.{message.customer_id}.created"
        body = json.dumps(asdict(message)).encode()

        msg = aio_pika.Message(
            body=body,
            delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
            content_type="application/json",
            message_id=f"{message.order_id}-{int(datetime.now().timestamp() * 1000)}",
        )
        await self.exchange.publish(msg, routing_key=routing_key)

    async def close(self) -> None:
        if self.connection:
            await self.connection.close()


# Usage
import asyncio

async def main():
    producer = RabbitMQProducer()
    await producer.connect("amqp://guest:guest@localhost")

    await producer.publish_order(OrderMessage(
        order_id="ORD-12345",
        customer_id="CUST-789",
        total=149.99,
        timestamp=datetime.now().isoformat(),
    ))

    await producer.close()

asyncio.run(main())

// RabbitMQProducer.cs
using RabbitMQ.Client;
using System.Text;
using System.Text.Json;

public record OrderMessage(string OrderId, string CustomerId, double Total, DateTime Timestamp);

public class RabbitMQProducer : IDisposable
{
    private IConnection? _connection;
    private IModel? _channel;

    public void Connect(string url = "amqp://localhost")
    {
        var factory = new ConnectionFactory { Uri = new Uri(url) };
        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();

        // Declare exchange (idempotent)
        _channel.ExchangeDeclare("orders", ExchangeType.Topic, durable: true);
    }

    public bool PublishOrder(OrderMessage message)
    {
        if (_channel == null) throw new InvalidOperationException("Producer not connected");

        var routingKey = $"orders.{message.CustomerId}.created";
        var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(message));

        var props = _channel.CreateBasicProperties();
        props.Persistent = true;
        props.ContentType = "application/json";
        props.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds());
        props.MessageId = $"{message.OrderId}-{DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()}";

        _channel.BasicPublish("orders", routingKey, props, body);
        return true;
    }

    public void Dispose()
    {
        _channel?.Close();
        _connection?.Close();
    }
}

// Usage
using var producer = new RabbitMQProducer();
producer.Connect("amqp://guest:guest@localhost");
producer.PublishOrder(new OrderMessage("ORD-12345", "CUST-789", 149.99, DateTime.UtcNow));

import amqp, { Message } from 'amqplib';

interface OrderMessage {
  orderId: string;
  customerId: string;
  total: number;
  timestamp: Date;
}

interface ProcessorConfig {
  exchangeName: string;
  queueName: string;
  routingPatterns: string[];
  prefetchCount: number;
  maxRetries: number;
}

class RabbitMQConsumer {
  private connection: amqp.Connection | null = null;
  private channel: amqp.Channel | null = null;
  private config: ProcessorConfig;

  constructor(config: ProcessorConfig) {
    this.config = config;
  }

  async connect(url: string = 'amqp://localhost'): Promise<void> {
    this.connection = await amqp.connect(url);
    this.channel = await this.connection.createChannel();

    // Set prefetch (QoS): process one message at a time
    await this.channel.prefetch(this.config.prefetchCount);

    // Declare exchange and queue
    await this.channel.assertExchange(this.config.exchangeName, 'topic', {
      durable: true,
    });

    await this.channel.assertQueue(this.config.queueName, {
      durable: true,
      arguments: {
        'x-dead-letter-exchange': `${this.config.exchangeName}-dlx`,
      },
    });

    // Bind queue to exchange with routing patterns
    for (const pattern of this.config.routingPatterns) {
      await this.channel.bindQueue(
        this.config.queueName,
        this.config.exchangeName,
        pattern
      );
    }
  }

  async consume(
    handler: (message: OrderMessage) => Promise<void>
  ): Promise<void> {
    if (!this.channel) {
      throw new Error('Consumer not connected');
    }

    await this.channel.consume(
      this.config.queueName,
      async (msg: Message | null) => {
        if (!msg) return;

        try {
          const content = JSON.parse(msg.content.toString()) as OrderMessage;
          const retries = msg.properties.headers['x-death']?.length ?? 0;

          if (retries > this.config.maxRetries) {
            console.error(
              `Message ${msg.properties.messageId} exceeded max retries. Moving to DLQ.`
            );
            // Acknowledge to remove from queue; DLQ will capture it
            this.channel!.ack(msg);
            return;
          }

          // Process with timeout
          await Promise.race([
            handler(content),
            new Promise((_, reject) =>
              setTimeout(
                () => reject(new Error('Processing timeout')),
                30000 // 30 second timeout
              )
            ),
          ]);

          // Acknowledge successful processing
          this.channel!.ack(msg);
          console.log(
            `Processed order ${content.orderId} successfully`
          );
        } catch (error) {
          console.error(
            `Error processing message ${msg.properties.messageId}:`,
            error
          );

          // Negative acknowledgment: requeue for retry
          this.channel!.nack(msg, false, true);
        }
      },
      { noAck: false } // Manual acknowledgment mode
    );

    console.log(`Consumer listening on queue: ${this.config.queueName}`);
  }

  async close(): Promise<void> {
    if (this.channel) {
      await this.channel.close();
    }
    if (this.connection) {
      await this.connection.close();
    }
  }
}

// Usage
const consumer = new RabbitMQConsumer({
  exchangeName: 'orders',
  queueName: 'order-processing-queue',
  routingPatterns: ['orders.*.created'],
  prefetchCount: 1,
  maxRetries: 3,
});

await consumer.connect('amqp://guest:guest@localhost');

await consumer.consume(async (order: OrderMessage) => {
  // Simulate processing
  console.log(`Processing order ${order.orderId} for customer ${order.customerId}`);
  
  if (order.total < 0) {
    throw new Error('Invalid order amount');
  }

  // Actual processing logic here
  await new Promise((resolve) => setTimeout(resolve, 1000));
});

// RabbitMQConsumer.java
import com.rabbitmq.client.*;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

public class RabbitMQConsumer {

    public record ProcessorConfig(
        String exchangeName, String queueName,
        List<String> routingPatterns, int prefetchCount, int maxRetries) {}

    private Connection connection;
    private Channel channel;
    private final ProcessorConfig config;
    private static final ObjectMapper mapper = new ObjectMapper();

    public RabbitMQConsumer(ProcessorConfig config) { this.config = config; }

    public void connect(String url) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri(url);
        this.connection = factory.newConnection();
        this.channel = connection.createChannel();

        channel.basicQos(config.prefetchCount());

        channel.exchangeDeclare(config.exchangeName(), "topic", true);
        channel.queueDeclare(config.queueName(), true, false, false,
            Map.of("x-dead-letter-exchange", config.exchangeName() + "-dlx"));

        for (String pattern : config.routingPatterns()) {
            channel.queueBind(config.queueName(), config.exchangeName(), pattern);
        }
    }

    public void consume(Consumer<Map<String, Object>> handler) throws IOException {
        if (channel == null) throw new IllegalStateException("Consumer not connected");

        channel.basicConsume(config.queueName(), false, (consumerTag, delivery) -> {
            String messageId = delivery.getProperties().getMessageId();
            try {
                @SuppressWarnings("unchecked")
                Map<String, Object> content = mapper.readValue(
                    new String(delivery.getBody(), StandardCharsets.UTF_8), Map.class);

                // Check retries via x-death header
                Map<String, Object> headers = delivery.getProperties().getHeaders();
                int retries = headers != null && headers.containsKey("x-death")
                    ? ((List<?>) headers.get("x-death")).size() : 0;

                if (retries > config.maxRetries()) {
                    System.err.println("Message " + messageId + " exceeded max retries. Moving to DLQ.");
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    return;
                }

                handler.accept(content);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println("Processed order " + content.get("orderId") + " successfully");

            } catch (Exception e) {
                System.err.println("Error processing message " + messageId + ": " + e.getMessage());
                try {
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                } catch (IOException ex) { ex.printStackTrace(); }
            }
        }, consumerTag -> {});

        System.out.println("Consumer listening on queue: " + config.queueName());
    }

    public void close() throws IOException, TimeoutException {
        if (channel != null) channel.close();
        if (connection != null) connection.close();
    }
}

# rabbitmq_consumer.py
import aio_pika
import json
import asyncio
from dataclasses import dataclass
from typing import Callable, Awaitable


@dataclass
class ProcessorConfig:
    exchange_name: str
    queue_name: str
    routing_patterns: list[str]
    prefetch_count: int
    max_retries: int


class RabbitMQConsumer:
    def __init__(self, config: ProcessorConfig):
        self.config = config
        self.connection = None
        self.channel = None

    async def connect(self, url: str = "amqp://localhost") -> None:
        self.connection = await aio_pika.connect_robust(url)
        self.channel = await self.connection.channel()

        await self.channel.set_qos(prefetch_count=self.config.prefetch_count)

        exchange = await self.channel.declare_exchange(
            self.config.exchange_name, aio_pika.ExchangeType.TOPIC, durable=True
        )

        queue = await self.channel.declare_queue(
            self.config.queue_name,
            durable=True,
            arguments={"x-dead-letter-exchange": f"{self.config.exchange_name}-dlx"},
        )

        for pattern in self.config.routing_patterns:
            await queue.bind(exchange, routing_key=pattern)

        self.queue = queue

    async def consume(self, handler: Callable[[dict], Awaitable[None]]) -> None:
        if not self.channel:
            raise RuntimeError("Consumer not connected")

        async def process_message(message: aio_pika.IncomingMessage):
            message_id = message.message_id
            async with message.process(requeue=True):
                try:
                    content = json.loads(message.body.decode())

                    # Check retries
                    headers = message.headers or {}
                    retries = len(headers.get("x-death", [])) if "x-death" in headers else 0

                    if retries > self.config.max_retries:
                        print(f"Message {message_id} exceeded max retries. Moving to DLQ.")
                        return

                    await asyncio.wait_for(handler(content), timeout=30)
                    print(f"Processed order {content.get('order_id')} successfully")

                except asyncio.TimeoutError:
                    print(f"Processing timeout for message {message_id}")
                    raise
                except Exception as e:
                    print(f"Error processing message {message_id}: {e}")
                    raise

        await self.queue.consume(process_message)
        print(f"Consumer listening on queue: {self.config.queue_name}")

    async def close(self) -> None:
        if self.connection:
            await self.connection.close()


# Usage
async def main():
    consumer = RabbitMQConsumer(ProcessorConfig(
        exchange_name="orders",
        queue_name="order-processing-queue",
        routing_patterns=["orders.*.created"],
        prefetch_count=1,
        max_retries=3,
    ))

    await consumer.connect("amqp://guest:guest@localhost")

    async def handle_order(order: dict):
        print(f"Processing order {order['order_id']} for customer {order['customer_id']}")
        if order["total"] < 0:
            raise ValueError("Invalid order amount")
        await asyncio.sleep(1)

    await consumer.consume(handle_order)
    await asyncio.Future()  # Run forever

asyncio.run(main())

// RabbitMQConsumer.cs
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;

public record ProcessorConfig(
    string ExchangeName, string QueueName,
    string[] RoutingPatterns, int PrefetchCount, int MaxRetries);

public class RabbitMQConsumer : IDisposable
{
    private IConnection? _connection;
    private IModel? _channel;
    private readonly ProcessorConfig _config;

    public RabbitMQConsumer(ProcessorConfig config) { _config = config; }

    public void Connect(string url = "amqp://localhost")
    {
        var factory = new ConnectionFactory { Uri = new Uri(url) };
        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();

        _channel.BasicQos(0, (ushort)_config.PrefetchCount, false);

        _channel.ExchangeDeclare(_config.ExchangeName, ExchangeType.Topic, durable: true);
        _channel.QueueDeclare(_config.QueueName, durable: true, exclusive: false, autoDelete: false,
            arguments: new Dictionary<string, object>
            {
                ["x-dead-letter-exchange"] = $"{_config.ExchangeName}-dlx"
            });

        foreach (var pattern in _config.RoutingPatterns)
            _channel.QueueBind(_config.QueueName, _config.ExchangeName, pattern);
    }

    public void Consume(Func<Dictionary<string, object>, Task> handler)
    {
        if (_channel == null) throw new InvalidOperationException("Consumer not connected");

        var consumer = new AsyncEventingBasicConsumer(_channel);
        consumer.Received += async (_, ea) =>
        {
            var messageId = ea.BasicProperties.MessageId;
            try
            {
                var content = JsonSerializer.Deserialize<Dictionary<string, object>>(
                    Encoding.UTF8.GetString(ea.Body.ToArray()))!;

                var headers = ea.BasicProperties.Headers;
                int retries = headers != null && headers.TryGetValue("x-death", out var xDeath)
                    ? (xDeath as List<object>)?.Count ?? 0 : 0;

                if (retries > _config.MaxRetries)
                {
                    Console.Error.WriteLine($"Message {messageId} exceeded max retries. Moving to DLQ.");
                    _channel.BasicAck(ea.DeliveryTag, false);
                    return;
                }

                using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
                await handler(content).WaitAsync(cts.Token);

                _channel.BasicAck(ea.DeliveryTag, false);
                Console.WriteLine($"Processed order {content.GetValueOrDefault("orderId")} successfully");
            }
            catch (Exception e)
            {
                Console.Error.WriteLine($"Error processing message {messageId}: {e.Message}");
                _channel.BasicNack(ea.DeliveryTag, false, requeue: true);
            }
        };

        _channel.BasicConsume(_config.QueueName, autoAck: false, consumer: consumer);
        Console.WriteLine($"Consumer listening on queue: {_config.QueueName}");
    }

    public void Dispose()
    {
        _channel?.Close();
        _connection?.Close();
    }
}

import { SQSClient, SendMessageCommand, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs';

interface OrderPayload {
  orderId: string;
  amount: number;
}

class SQSMessageQueue {
  private client: SQSClient;
  private queueUrl: string;

  constructor(region: string, queueUrl: string) {
    this.client = new SQSClient({ region });
    this.queueUrl = queueUrl;
  }

  async sendMessage(payload: OrderPayload, delaySeconds: number = 0): Promise<string> {
    const command = new SendMessageCommand({
      QueueUrl: this.queueUrl,
      MessageBody: JSON.stringify(payload),
      DelaySeconds: delaySeconds,
      MessageAttributes: {
        'OrderId': {
          StringValue: payload.orderId,
          DataType: 'String',
        },
        'Timestamp': {
          StringValue: new Date().toISOString(),
          DataType: 'String',
        },
      },
    });

    const response = await this.client.send(command);
    return response.MessageId!;
  }

  async receiveMessages(maxMessages: number = 10, visibilityTimeout: number = 30) {
    const command = new ReceiveMessageCommand({
      QueueUrl: this.queueUrl,
      MaxNumberOfMessages: maxMessages,
      VisibilityTimeout: visibilityTimeout,
      WaitTimeSeconds: 20, // Long polling
      MessageAttributeNames: ['All'],
    });

    const response = await this.client.send(command);
    return response.Messages || [];
  }

  async deleteMessage(receiptHandle: string): Promise<void> {
    const command = new DeleteMessageCommand({
      QueueUrl: this.queueUrl,
      ReceiptHandle: receiptHandle,
    });

    await this.client.send(command);
  }
}

// Consumer implementation
class SQSConsumer {
  private queue: SQSMessageQueue;
  private isRunning: boolean = false;

  constructor(queue: SQSMessageQueue) {
    this.queue = queue;
  }

  async start(handler: (payload: OrderPayload) => Promise<void>): Promise<void> {
    this.isRunning = true;

    while (this.isRunning) {
      try {
        const messages = await this.queue.receiveMessages(10, 60);

        if (messages.length === 0) {
          continue;
        }

        for (const message of messages) {
          try {
            const payload = JSON.parse(message.Body!) as OrderPayload;

            // Process with timeout
            await Promise.race([
              handler(payload),
              new Promise((_, reject) =>
                setTimeout(
                  () => reject(new Error('Processing timeout')),
                  50000 // 50 second timeout (less than 60 second visibility)
                )
              ),
            ]);

            // Delete message on success
            await this.queue.deleteMessage(message.ReceiptHandle!);
            console.log(`Processed order ${payload.orderId}`);
          } catch (error) {
            console.error(
              `Error processing message ${message.MessageId}:`,
              error
            );
            // Don't delete; message will reappear after visibility timeout
          }
        }
      } catch (error) {
        console.error('Error in consumer loop:', error);
        await new Promise((resolve) => setTimeout(resolve, 5000));
      }
    }
  }

  stop(): void {
    this.isRunning = false;
  }
}

// SQSMessageQueue.java
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.*;
import software.amazon.awssdk.regions.Region;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.List;
import java.util.Map;

public class SQSMessageQueue {
    private final SqsClient client;
    private final String queueUrl;
    private static final ObjectMapper mapper = new ObjectMapper();

    public record OrderPayload(String orderId, double amount) {}

    public SQSMessageQueue(String region, String queueUrl) {
        this.client = SqsClient.builder().region(Region.of(region)).build();
        this.queueUrl = queueUrl;
    }

    public String sendMessage(OrderPayload payload, int delaySeconds) throws Exception {
        SendMessageResponse response = client.sendMessage(SendMessageRequest.builder()
            .queueUrl(queueUrl)
            .messageBody(mapper.writeValueAsString(payload))
            .delaySeconds(delaySeconds)
            .messageAttributes(Map.of(
                "OrderId", MessageAttributeValue.builder().stringValue(payload.orderId()).dataType("String").build(),
                "Timestamp", MessageAttributeValue.builder().stringValue(java.time.Instant.now().toString()).dataType("String").build()
            ))
            .build());
        return response.messageId();
    }

    public List<Message> receiveMessages(int maxMessages, int visibilityTimeout) {
        ReceiveMessageResponse response = client.receiveMessage(ReceiveMessageRequest.builder()
            .queueUrl(queueUrl)
            .maxNumberOfMessages(maxMessages)
            .visibilityTimeout(visibilityTimeout)
            .waitTimeSeconds(20) // Long polling
            .messageAttributeNames("All")
            .build());
        return response.messages();
    }

    public void deleteMessage(String receiptHandle) {
        client.deleteMessage(DeleteMessageRequest.builder()
            .queueUrl(queueUrl)
            .receiptHandle(receiptHandle)
            .build());
    }
}

// SQSConsumer.java
class SQSConsumer {
    private final SQSMessageQueue queue;
    private volatile boolean isRunning = false;

    public SQSConsumer(SQSMessageQueue queue) { this.queue = queue; }

    public void start(java.util.function.Consumer<SQSMessageQueue.OrderPayload> handler) {
        isRunning = true;

        while (isRunning) {
            try {
                List<Message> messages = queue.receiveMessages(10, 60);

                for (Message message : messages) {
                    try {
                        SQSMessageQueue.OrderPayload payload = new ObjectMapper()
                            .readValue(message.body(), SQSMessageQueue.OrderPayload.class);
                        handler.accept(payload);
                        queue.deleteMessage(message.receiptHandle());
                        System.out.println("Processed order " + payload.orderId());
                    } catch (Exception e) {
                        System.err.println("Error processing message " + message.messageId() + ": " + e.getMessage());
                        // Don't delete; message will reappear after visibility timeout
                    }
                }
            } catch (Exception e) {
                System.err.println("Error in consumer loop: " + e.getMessage());
                try { Thread.sleep(5000); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
            }
        }
    }

    public void stop() { isRunning = false; }
}

# sqs_message_queue.py
import boto3
import json
import asyncio
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable, Awaitable


@dataclass
class OrderPayload:
    order_id: str
    amount: float


class SQSMessageQueue:
    def __init__(self, region: str, queue_url: str):
        self.client = boto3.client("sqs", region_name=region)
        self.queue_url = queue_url

    def send_message(self, payload: OrderPayload, delay_seconds: int = 0) -> str:
        response = self.client.send_message(
            QueueUrl=self.queue_url,
            MessageBody=json.dumps({"order_id": payload.order_id, "amount": payload.amount}),
            DelaySeconds=delay_seconds,
            MessageAttributes={
                "OrderId": {"StringValue": payload.order_id, "DataType": "String"},
                "Timestamp": {"StringValue": datetime.now(timezone.utc).isoformat(), "DataType": "String"},
            },
        )
        return response["MessageId"]

    def receive_messages(self, max_messages: int = 10, visibility_timeout: int = 30) -> list:
        response = self.client.receive_message(
            QueueUrl=self.queue_url,
            MaxNumberOfMessages=max_messages,
            VisibilityTimeout=visibility_timeout,
            WaitTimeSeconds=20,  # Long polling
            MessageAttributeNames=["All"],
        )
        return response.get("Messages", [])

    def delete_message(self, receipt_handle: str) -> None:
        self.client.delete_message(QueueUrl=self.queue_url, ReceiptHandle=receipt_handle)


class SQSConsumer:
    def __init__(self, queue: SQSMessageQueue):
        self.queue = queue
        self.is_running = False

    async def start(self, handler: Callable[[OrderPayload], Awaitable[None]]) -> None:
        self.is_running = True

        while self.is_running:
            try:
                messages = self.queue.receive_messages(10, 60)

                for message in messages:
                    try:
                        data = json.loads(message["Body"])
                        payload = OrderPayload(order_id=data["order_id"], amount=data["amount"])

                        await asyncio.wait_for(handler(payload), timeout=50)

                        self.queue.delete_message(message["ReceiptHandle"])
                        print(f"Processed order {payload.order_id}")
                    except Exception as e:
                        print(f"Error processing message {message['MessageId']}: {e}")
                        # Don't delete; message will reappear after visibility timeout

            except Exception as e:
                print(f"Error in consumer loop: {e}")
                await asyncio.sleep(5)

    def stop(self) -> None:
        self.is_running = False

// SQSMessageQueue.cs
using Amazon.SQS;
using Amazon.SQS.Model;
using System.Text.Json;

public record OrderPayload(string OrderId, double Amount);

public class SQSMessageQueue
{
    private readonly AmazonSQSClient _client;
    private readonly string _queueUrl;

    public SQSMessageQueue(string region, string queueUrl)
    {
        _client = new AmazonSQSClient(Amazon.RegionEndpoint.GetBySystemName(region));
        _queueUrl = queueUrl;
    }

    public async Task<string> SendMessageAsync(OrderPayload payload, int delaySeconds = 0)
    {
        var response = await _client.SendMessageAsync(new SendMessageRequest
        {
            QueueUrl = _queueUrl,
            MessageBody = JsonSerializer.Serialize(payload),
            DelaySeconds = delaySeconds,
            MessageAttributes = new Dictionary<string, MessageAttributeValue>
            {
                ["OrderId"] = new() { StringValue = payload.OrderId, DataType = "String" },
                ["Timestamp"] = new() { StringValue = DateTime.UtcNow.ToString("O"), DataType = "String" },
            }
        });
        return response.MessageId;
    }

    public async Task<List<Message>> ReceiveMessagesAsync(int maxMessages = 10, int visibilityTimeout = 30)
    {
        var response = await _client.ReceiveMessageAsync(new ReceiveMessageRequest
        {
            QueueUrl = _queueUrl,
            MaxNumberOfMessages = maxMessages,
            VisibilityTimeout = visibilityTimeout,
            WaitTimeSeconds = 20,
            MessageAttributeNames = new List<string> { "All" },
        });
        return response.Messages;
    }

    public async Task DeleteMessageAsync(string receiptHandle) =>
        await _client.DeleteMessageAsync(_queueUrl, receiptHandle);
}

public class SQSConsumer
{
    private readonly SQSMessageQueue _queue;
    private bool _isRunning;

    public SQSConsumer(SQSMessageQueue queue) { _queue = queue; }

    public async Task StartAsync(Func<OrderPayload, Task> handler)
    {
        _isRunning = true;

        while (_isRunning)
        {
            try
            {
                var messages = await _queue.ReceiveMessagesAsync(10, 60);

                foreach (var message in messages)
                {
                    try
                    {
                        var payload = JsonSerializer.Deserialize<OrderPayload>(message.Body)!;

                        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(50));
                        await handler(payload).WaitAsync(cts.Token);

                        await _queue.DeleteMessageAsync(message.ReceiptHandle);
                        Console.WriteLine($"Processed order {payload.OrderId}");
                    }
                    catch (Exception e)
                    {
                        Console.Error.WriteLine($"Error processing message {message.MessageId}: {e.Message}");
                        // Don't delete; message will reappear after visibility timeout
                    }
                }
            }
            catch (Exception e)
            {
                Console.Error.WriteLine($"Error in consumer loop: {e.Message}");
                await Task.Delay(5000);
            }
        }
    }

    public void Stop() => _isRunning = false;
}

// Producer
const queue = new SQSMessageQueue('us-east-1', 'https://sqs.us-east-1.amazonaws.com/123456789/task-queue');

for (let i = 0; i < 100; i++) {
  await queue.sendMessage({
    orderId: `TASK-${i}`,
    amount: Math.random() * 1000,
  });
}

// Multiple consumers
const consumer = new SQSConsumer(queue);

const handler = async (payload: OrderPayload) => {
  console.log(`Processing ${payload.orderId}`);
  await new Promise((resolve) => setTimeout(resolve, Math.random() * 5000));
};

// Start 5 parallel consumers
Promise.all([...Array(5)].map(() => consumer.start(handler)));

// Work Queue Pattern - Producer
SQSMessageQueue queue = new SQSMessageQueue("us-east-1",
    "https://sqs.us-east-1.amazonaws.com/123456789/task-queue");

for (int i = 0; i < 100; i++) {
    queue.sendMessage(new SQSMessageQueue.OrderPayload("TASK-" + i, Math.random() * 1000), 0);
}

// Work Queue Pattern - Multiple consumers via thread pool
SQSConsumer consumer = new SQSConsumer(queue);

java.util.function.Consumer<SQSMessageQueue.OrderPayload> handler = payload -> {
    System.out.println("Processing " + payload.orderId());
    try { Thread.sleep((long)(Math.random() * 5000)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
};

// Start 5 parallel consumers
java.util.concurrent.ExecutorService executor = java.util.concurrent.Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
    executor.submit(() -> consumer.start(handler));
}

# Work Queue Pattern - Producer
import asyncio
import random

queue = SQSMessageQueue("us-east-1",
    "https://sqs.us-east-1.amazonaws.com/123456789/task-queue")

for i in range(100):
    queue.send_message(OrderPayload(order_id=f"TASK-{i}", amount=random.random() * 1000))

# Multiple consumers
consumer = SQSConsumer(queue)

async def handler(payload: OrderPayload):
    print(f"Processing {payload.order_id}")
    await asyncio.sleep(random.random() * 5)

# Start 5 parallel consumers
async def main():
    await asyncio.gather(*[consumer.start(handler) for _ in range(5)])

asyncio.run(main())

// Work Queue Pattern - Producer
var queue = new SQSMessageQueue("us-east-1",
    "https://sqs.us-east-1.amazonaws.com/123456789/task-queue");

for (int i = 0; i < 100; i++)
    await queue.SendMessageAsync(new OrderPayload($"TASK-{i}", Random.Shared.NextDouble() * 1000));

// Multiple consumers
var consumer = new SQSConsumer(queue);

Func<OrderPayload, Task> handler = async payload =>
{
    Console.WriteLine($"Processing {payload.OrderId}");
    await Task.Delay((int)(Random.Shared.NextDouble() * 5000));
};

// Start 5 parallel consumers
await Task.WhenAll(Enumerable.Range(0, 5).Select(_ => consumer.StartAsync(handler)));

// In RabbitMQ, create fanout exchange
async function setupFanout(channel: amqp.Channel) {
  const exchangeName = 'user-events-fanout';
  
  // Fanout: broadcast to all bound queues
  await channel.assertExchange(exchangeName, 'fanout', { durable: true });

  // Email service queue
  await channel.assertQueue('email-queue', { durable: true });
  await channel.bindQueue('email-queue', exchangeName, '');

  // Analytics queue
  await channel.assertQueue('analytics-queue', { durable: true });
  await channel.bindQueue('analytics-queue', exchangeName, '');

  // Notification queue
  await channel.assertQueue('notification-queue', { durable: true });
  await channel.bindQueue('notification-queue', exchangeName, '');

  return exchangeName;
}

// Single user registration message triggers three consumers
const exchangeName = await setupFanout(channel);
await channel.publish(
  exchangeName,
  '', // Fanout ignores routing key
  Buffer.from(JSON.stringify({ userId: 'user-123', email: 'user@example.com' })),
  { persistent: true }
);

// Pub/Sub Fanout Pattern
import com.rabbitmq.client.Channel;
import com.fasterxml.jackson.databind.ObjectMapper;

public String setupFanout(Channel channel) throws Exception {
    String exchangeName = "user-events-fanout";

    // Fanout: broadcast to all bound queues
    channel.exchangeDeclare(exchangeName, "fanout", true);

    channel.queueDeclare("email-queue", true, false, false, null);
    channel.queueBind("email-queue", exchangeName, "");

    channel.queueDeclare("analytics-queue", true, false, false, null);
    channel.queueBind("analytics-queue", exchangeName, "");

    channel.queueDeclare("notification-queue", true, false, false, null);
    channel.queueBind("notification-queue", exchangeName, "");

    return exchangeName;
}

// Publish fanout message
String exchangeName = setupFanout(channel);
byte[] body = new ObjectMapper().writeValueAsBytes(
    java.util.Map.of("userId", "user-123", "email", "user@example.com")
);
com.rabbitmq.client.AMQP.BasicProperties props =
    new com.rabbitmq.client.AMQP.BasicProperties.Builder().deliveryMode(2).build();
channel.basicPublish(exchangeName, "", props, body); // Fanout ignores routing key

# Pub/Sub Fanout Pattern
import aio_pika
import json


async def setup_fanout(channel: aio_pika.Channel) -> aio_pika.Exchange:
    exchange_name = "user-events-fanout"

    # Fanout: broadcast to all bound queues
    exchange = await channel.declare_exchange(
        exchange_name, aio_pika.ExchangeType.FANOUT, durable=True
    )

    for queue_name in ("email-queue", "analytics-queue", "notification-queue"):
        queue = await channel.declare_queue(queue_name, durable=True)
        await queue.bind(exchange, routing_key="")  # Fanout ignores routing key

    return exchange


# Publish fanout message
exchange = await setup_fanout(channel)
await exchange.publish(
    aio_pika.Message(
        body=json.dumps({"user_id": "user-123", "email": "user@example.com"}).encode(),
        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
    ),
    routing_key="",  # Fanout ignores routing key
)

// Pub/Sub Fanout Pattern
using RabbitMQ.Client;
using System.Text;
using System.Text.Json;

static string SetupFanout(IModel channel)
{
    const string exchangeName = "user-events-fanout";

    // Fanout: broadcast to all bound queues
    channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout, durable: true);

    foreach (var queueName in new[] { "email-queue", "analytics-queue", "notification-queue" })
    {
        channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);
        channel.QueueBind(queueName, exchangeName, routingKey: ""); // Fanout ignores routing key
    }

    return exchangeName;
}

// Publish fanout message
var exchangeName = SetupFanout(channel);
var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new { userId = "user-123", email = "user@example.com" }));
var props = channel.CreateBasicProperties();
props.Persistent = true;
channel.BasicPublish(exchangeName, routingKey: "", props, body); // Fanout ignores routing key

async function setupTopicRouting(channel: amqp.Channel) {
  const exchangeName = 'orders-topic';
  
  await channel.assertExchange(exchangeName, 'topic', { durable: true });

  // Queue for all order updates
  await channel.assertQueue('all-orders-queue', { durable: true });
  await channel.bindQueue('all-orders-queue', exchangeName, 'orders.#'); // All events

  // Queue for urgent orders only
  await channel.assertQueue('urgent-orders-queue', { durable: true });
  await channel.bindQueue('urgent-orders-queue', exchangeName, 'orders.*.urgent');

  // Queue for processing orders
  await channel.assertQueue('processing-queue', { durable: true });
  await channel.bindQueue('processing-queue', exchangeName, 'orders.processing.*');

  return exchangeName;
}

// Publish with topic routing
await channel.publish(
  'orders-topic',
  'orders.processing.urgent',
  Buffer.from(JSON.stringify({ orderId: '12345', priority: 'urgent' })),
  { persistent: true }
);

// 'urgent-orders-queue' and 'processing-queue' both receive this message
// 'all-orders-queue' also receives it

// Topic-Based Routing Pattern
import com.rabbitmq.client.Channel;
import com.fasterxml.jackson.databind.ObjectMapper;

public String setupTopicRouting(Channel channel) throws Exception {
    String exchangeName = "orders-topic";

    channel.exchangeDeclare(exchangeName, "topic", true);

    channel.queueDeclare("all-orders-queue", true, false, false, null);
    channel.queueBind("all-orders-queue", exchangeName, "orders.#"); // All events

    channel.queueDeclare("urgent-orders-queue", true, false, false, null);
    channel.queueBind("urgent-orders-queue", exchangeName, "orders.*.urgent");

    channel.queueDeclare("processing-queue", true, false, false, null);
    channel.queueBind("processing-queue", exchangeName, "orders.processing.*");

    return exchangeName;
}

// Publish with topic routing
byte[] body = new ObjectMapper().writeValueAsBytes(
    java.util.Map.of("orderId", "12345", "priority", "urgent")
);
com.rabbitmq.client.AMQP.BasicProperties props =
    new com.rabbitmq.client.AMQP.BasicProperties.Builder().deliveryMode(2).build();
channel.basicPublish("orders-topic", "orders.processing.urgent", props, body);
// 'urgent-orders-queue' and 'processing-queue' both receive this
// 'all-orders-queue' also receives it

# Topic-Based Routing Pattern
import aio_pika
import json


async def setup_topic_routing(channel: aio_pika.Channel) -> aio_pika.Exchange:
    exchange_name = "orders-topic"

    exchange = await channel.declare_exchange(
        exchange_name, aio_pika.ExchangeType.TOPIC, durable=True
    )

    bindings = [
        ("all-orders-queue", "orders.#"),       # All events
        ("urgent-orders-queue", "orders.*.urgent"),
        ("processing-queue", "orders.processing.*"),
    ]

    for queue_name, routing_key in bindings:
        queue = await channel.declare_queue(queue_name, durable=True)
        await queue.bind(exchange, routing_key=routing_key)

    return exchange


# Publish with topic routing
exchange = await setup_topic_routing(channel)
await exchange.publish(
    aio_pika.Message(
        body=json.dumps({"order_id": "12345", "priority": "urgent"}).encode(),
        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
    ),
    routing_key="orders.processing.urgent",
)
# 'urgent-orders-queue' and 'processing-queue' both receive this
# 'all-orders-queue' also receives it

// Topic-Based Routing Pattern
using RabbitMQ.Client;
using System.Text;
using System.Text.Json;

static string SetupTopicRouting(IModel channel)
{
    const string exchangeName = "orders-topic";

    channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, durable: true);

    var bindings = new[]
    {
        ("all-orders-queue",      "orders.#"),           // All events
        ("urgent-orders-queue",   "orders.*.urgent"),
        ("processing-queue",      "orders.processing.*"),
    };

    foreach (var (queueName, routingKey) in bindings)
    {
        channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);
        channel.QueueBind(queueName, exchangeName, routingKey);
    }

    return exchangeName;
}

// Publish with topic routing
var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new { orderId = "12345", priority = "urgent" }));
var props = channel.CreateBasicProperties();
props.Persistent = true;
channel.BasicPublish("orders-topic", "orders.processing.urgent", props, body);
// 'urgent-orders-queue' and 'processing-queue' both receive this
// 'all-orders-queue' also receives it