rabbitmq-delayed-message.js

/*
* This code requires both RabbitMQ and delayed message plugin is installed.
* The plugin can be found here https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/
*
* Refeence: https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/
*/

'use strict';

var amqp = require('amqplib'),
    connection = amqp.connect('amqp://localhost'),
    queue = 'my-messages',
    exchange = 'my-exchange';
    

connection.then(function (conn) {

    return conn.createChannel()
    .then(function (ch) {

        // Creates a queue
        ch.assertQueue(queue).then(function () {

            // Creates an exchange
            ch.assertExchange(exchange, 'x-delayed-message', { arguments: { 'x-delayed-type': 'direct' } }).then(function () {

                // Creates a bind between queue and exchange
                ch.bindQueue(queue, exchange).then(function () {

                    // Send messages with delay
                    ch.publish(exchange, '', new Buffer('3. my delayed message'), { persistent: true, headers: { 'x-delay': 10000 } });
                    ch.publish(exchange, '', new Buffer('1. my delayed message'), { persistent: true, headers: { 'x-delay': 3000 } });
                    ch.publish(exchange, '', new Buffer('4. my delayed message'), { persistent: true, headers: { 'x-delay': 12000 } });
                    ch.publish(exchange, '', new Buffer('2. my delayed message'), { persistent: true, headers: { 'x-delay': 5000 } });

                    console.log('messages sent!');
                });
            });
        });
    });
});

connection.then(function (conn) {

    return conn.createChannel()
    .then(function (ch) {

        ch.consume(queue, function (msg) {
            if (msg !== null) {
                console.log(msg.content.toString());
                ch.ack(msg);
            }
        });
    });
});