RabbitMQ with Node.js: Complete Guide for Microservices

// consumer.js — รับและประมวลผลข้อความจาก RabbitMQ

const amqp = require('amqplib/callback_api');

const QUEUE_NAME = 'tasks';
const RABBITMQ_URL = process.env.RABBITMQ_URL || 'amqp://guest:guest@localhost:5672';

// จำลองการส่งอีเมล
function sendEmail(payload) {
    return new Promise((resolve, reject) => {
        console.log(`📧 Email service: Sending to ${payload.to}...`);
        // Simulate email sending delay
        setTimeout(() => {
            console.log(`✅ Email sent to ${payload.to}`);
            resolve();
        }, 1000);
    });
}

function processMessage(message) {
    const messageData = JSON.parse(message.content.toString());
    console.log('📦 Message received:', JSON.stringify(messageData, null, 2));
    
    if (messageData.type === 'send_email') {
        return sendEmail(messageData.payload)
            .then(() => {
                console.log('✅ Email task completed');
            })
            .catch(err => {
                console.error('❌ Email task failed:', err.message);
                throw err;
            });
    }
    
    console.log('⚠️ Unknown message type:', messageData.type);
}

function startConsumer() {
    amqp.connect(RABBITMQ_URL, (err, connection) => {
        if (err) {
            console.error('❌ RabbitMQ connection error:', err.message);
            console.log('🔄 Retrying in 5 seconds...');
            setTimeout(startConsumer, 5000);
            return;
        }

        console.log('✅ Connected to RabbitMQ!');

        connection.createChannel((err, channel) => {
            if (err) {
                console.error('❌ Channel creation error:', err.message);
                return;
            }

            channel.assertQueue(QUEUE_NAME, {
                durable: true
            });

            // ตั้งค่าความเร็วในการประมวลผล — 1 ข้อความต่อครั้ง
            channel.prefetch(1);

            console.log('⏳ Waiting for messages in queue:', QUEUE_NAME);

            // Start consuming messages
            channel.consume(QUEUE_NAME, (msg) => {
                processMessage(msg)
                    .then(() => {
                        // Acknowledge after success
                        channel.ack(msg);
                    })
                    .catch(() => {
                        // Reject and requeue (หรือ dead-letter queue)
                        channel.nack(msg, false, true);
                    });
            }, {
                noAck: false  // ต้อง ack manually (เพื่อ reliability)
            });
        });
    });
}

// Run with auto-reconnect
startConsumer();

// Graceful shutdown
process.on('SIGINT', () => {
    console.log('\n👋 Shutting down consumer...');
    process.exit(0);
});