Message Queues & Event Streaming

from kafka import KafkaProducer, KafkaConsumer
import json

# ── PRODUCER: Order Service ──────────────────────────────────────────────
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def place_order(order_id, user_id, items):
    event = {
        "order_id": order_id,
        "user_id": user_id,
        "items": items,
        "status": "placed"
    }
    # topic: "order_placed" এ message publish করুন
    producer.send('order_placed', value=event)
    producer.flush()
    print(f"✅ Order {order_id} event published to Kafka")

place_order("ORD-001", "USR-42", [{"item": "Biryani", "qty": 2}])


# ── CONSUMER: Email Service ───────────────────────────────────────────────
consumer = KafkaConsumer(
    'order_placed',
    bootstrap_servers='localhost:9092',
    group_id='email-service-group',       # Consumer Group ID
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

print("📧 Email Service listening for order events...")
for message in consumer:
    order = message.value
    print(f"Sending confirmation email for order: {order['order_id']}")
    # send_email(order['user_id'], order['order_id'])
    # offset automatically commit হয় — message processed হলে

const amqp = require('amqplib');

// ── PUBLISHER ────────────────────────────────────────────────────────────
async function publishImageTask(imageUrl) {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queue = 'image_resize';
  await channel.assertQueue(queue, { durable: true }); // queue crash এ survive করে

  const task = JSON.stringify({ imageUrl, width: 800, height: 600 });
  channel.sendToQueue(queue, Buffer.from(task), { persistent: true });

  console.log("📤 Task queued:", imageUrl);
  await channel.close();
  await connection.close();
}

publishImageTask('https://cdn.example.com/photo-001.jpg');


// ── WORKER (Consumer) ─────────────────────────────────────────────────────
async function startWorker() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queue = 'image_resize';
  await channel.assertQueue(queue, { durable: true });
  channel.prefetch(1); // একটা message process শেষ না হলে পরেরটা নেবে না

  console.log("🔧 Worker listening for image tasks...");

  channel.consume(queue, async (msg) => {
    const task = JSON.parse(msg.content.toString());

    try {
      console.log(`Processing: ${task.imageUrl}`);
      await resizeImage(task.imageUrl, task.width, task.height); // actual work
      channel.ack(msg);          // ✅ সফল → broker কে জানাও, message delete হবে
      console.log("✅ Image processed successfully");
    } catch (error) {
      console.error("❌ Failed:", error.message);
      channel.nack(msg, false, false); // ❌ fail → requeue: false → DLQ তে যাবেন
    }
  });
}

startWorker();

const amqp = require('amqplib');

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

  // 1. Dead Letter Exchange তৈরি করুন
  await channel.assertExchange('dlx_exchange', 'direct', { durable: true });

  // 2. Dead Letter Queue তৈরি করুন
  await channel.assertQueue('dead_letter_queue', { durable: true });
  await channel.bindQueue('dead_letter_queue', 'dlx_exchange', 'failed');

  // 3. Main queue — DLX config সহ
  await channel.assertQueue('image_resize', {
    durable: true,
    arguments: {
      'x-dead-letter-exchange': 'dlx_exchange',  // fail হলে এখানে যাবেন
      'x-dead-letter-routing-key': 'failed',
      'x-message-ttl': 30000,                    // 30 sec এর মধ্যে process না হলে DLQ
    }
  });

  console.log("✅ DLQ setup complete");
  // এখন যেকোনো nack(msg, false, false) → dlx_exchange → dead_letter_queue
  await connection.close();
}

setupDLQ();