// Producer: send a message to a queue
async function enqueueEmail(emailData: EmailPayload) {
  await queue.send('email-queue', {
    to: emailData.to,
    subject: emailData.subject,
    body: emailData.body
  });
}

// Consumer 1: reads from the queue
queue.consume('email-queue', async (message) => {
  await sendEmail(message.to, message.subject, message.body);
  message.ack(); // Acknowledge: remove from queue
});

// Consumer 2: also reads from the same queue
// But each message goes to EITHER Consumer 1 OR Consumer 2, not both
queue.consume('email-queue', async (message) => {
  await sendEmail(message.to, message.subject, message.body);
  message.ack();
});

// Producer: publish to a topic
async function publishOrderEvent(order: Order) {
  await topic.publish('order-events', {
    type: 'OrderPlaced',
    data: { orderId: order.id, total: order.total, customerId: order.customerId }
  });
}

// Subscriber 1: Email service gets the event
topic.subscribe('order-events', 'email-service', async (message) => {
  await sendOrderConfirmation(message.data.customerId, message.data.orderId);
});

// Subscriber 2: Analytics service ALSO gets the same event
topic.subscribe('order-events', 'analytics-service', async (message) => {
  await trackPurchase(message.data.orderId, message.data.total);
});

// Subscriber 3: Inventory service ALSO gets it
topic.subscribe('order-events', 'inventory-service', async (message) => {
  await reserveStock(message.data.orderId);
});

Producer
   |
   v
[Topic: order-events]
   |         |          |
   v         v          v
[Queue:   [Queue:    [Queue:
 email]    analytics]  inventory]
   |         |          |
   v         v          v
Email     Analytics   Inventory
Service   Service     Service

// Fan-out setup (conceptual)
// 1. Create a topic
const topic = broker.createTopic('order-events');

// 2. Each service has its own queue subscribed to the topic
const emailQueue = broker.createQueue('email-processing');
const analyticsQueue = broker.createQueue('analytics-processing');
const inventoryQueue = broker.createQueue('inventory-processing');

topic.bindQueue(emailQueue);
topic.bindQueue(analyticsQueue);
topic.bindQueue(inventoryQueue);

// 3. Each service consumes from its own queue independently
// If analytics is slow, it doesn't block email or inventory
emailQueue.consume(async (msg) => {
  await sendConfirmation(msg.data);
  msg.ack();
});

// RabbitMQ-style routing with routing keys
// Publish with a routing key
await exchange.publish('orders.placed', orderEvent);
await exchange.publish('orders.cancelled', cancelEvent);
await exchange.publish('payments.completed', paymentEvent);

// Email service: only wants order events
emailQueue.bindTo(exchange, 'orders.*');

// Analytics service: wants everything
analyticsQueue.bindTo(exchange, '#');

// Refund service: only wants cancellations
refundQueue.bindTo(exchange, 'orders.cancelled');

// RabbitMQ: publish to an exchange with routing
import amqp from 'amqplib';

const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();

// Create a topic exchange
await channel.assertExchange('orders', 'topic', { durable: true });

// Publish with routing key
channel.publish('orders', 'order.placed', Buffer.from(JSON.stringify({
  orderId: 'ord-123',
  total: 59.98
})));

// Consumer: bind queue to exchange with pattern
await channel.assertQueue('email-notifications', { durable: true });
await channel.bindQueue('email-notifications', 'orders', 'order.*');

channel.consume('email-notifications', (msg) => {
  const event = JSON.parse(msg.content.toString());
  console.log('Processing:', event);
  channel.ack(msg);
});

// Kafka: produce and consume from a topic
import { Kafka } from 'kafkajs';

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'email-service' });

// Produce
await producer.connect();
await producer.send({
  topic: 'order-events',
  messages: [{
    key: 'ord-123',  // Partition key: same order always goes to same partition
    value: JSON.stringify({ type: 'OrderPlaced', orderId: 'ord-123', total: 59.98 })
  }]
});

// Consume
await consumer.connect();
await consumer.subscribe({ topic: 'order-events', fromBeginning: false });
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value.toString());
    console.log(`Partition ${partition}, Offset ${message.offset}:`, event);
  }
});

// SNS + SQS: fan-out with managed services
import { SNSClient, PublishCommand } from '@aws-sdk/client-sns';
import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs';

const sns = new SNSClient({});
const sqs = new SQSClient({});

// Publish to SNS topic (fans out to all subscribed queues)
await sns.send(new PublishCommand({
  TopicArn: 'arn:aws:sns:us-east-1:123456789:order-events',
  Message: JSON.stringify({ type: 'OrderPlaced', orderId: 'ord-123' }),
  MessageAttributes: {
    eventType: { DataType: 'String', StringValue: 'OrderPlaced' }
  }
}));

// Consume from SQS queue (subscribed to the SNS topic)
const response = await sqs.send(new ReceiveMessageCommand({
  QueueUrl: 'https://sqs.us-east-1.amazonaws.com/123456789/email-queue',
  MaxNumberOfMessages: 10,
  WaitTimeSeconds: 20  // Long polling
}));

for (const msg of response.Messages || []) {
  const event = JSON.parse(JSON.parse(msg.Body).Message);
  await processEvent(event);
  await sqs.send(new DeleteMessageCommand({
    QueueUrl: 'https://sqs.us-east-1.amazonaws.com/123456789/email-queue',
    ReceiptHandle: msg.ReceiptHandle
  }));
}