Testing RabbitMQ's ability to recover from a lot of channels and queues

stress.js

// npm install amqplib
var rmq = require ('amqplib/callback_api');

TOTAL = process.argv[2] || 5000;

// three local cluster nodes
var nodes = [25671, 25672, 25673];

var connected = latch(nodes.length, go);

var conns = [];
nodes.forEach(function(port, index) {
    rmq.connect('amqp://localhost:' + port, function(_, c) {
        conns[index] = c;
        connected();
    });
});

function latch(num, oncomplete) {
    return function() {
        num--;
        if (num == 0) oncomplete();
    };
}

function go() {

    var allchannels = [];
    
    function closeAll() {
        console.log('Closing ...');
        allchannels.forEach(function(ch) { ch.close(channelsClosed); });
    }

    var total = TOTAL;

    var queuesDone = latch(TOTAL, closeAll);
    var channelsClosed = latch(TOTAL, function() {
        console.log('Closed.'); conns.forEach(function(c) { c.close(); });
    });

    for (var i = 0; i < TOTAL; i++) {

        var queues = 0;
        var consumers = 0;
        var channels = 0;

        var conn = conns.shift();
        conns.push(conn);
        
        conn.createChannel(function(_, ch) {
            channels ++;
            allchannels.push(ch);
            if (channels % 1000 == 0) console.log('Created %d channels', channels);
            ch.assertQueue('', {autoDelete: true}, function(_, q1) {
                queues ++; if (queues % 1000 == 0) console.log('%d queues created', queues);
                ch.assertQueue('', {autoDelete: true}, function(_, q2) {
                    queues ++; if (queues % 1000 == 0) console.log('%d queues created', queues);
                    ch.consume(q1.queue, function() {}, {noAck: false}, function() {
                        consumers ++; if (consumers % 1000 == 0) console.log('%d consumers', consumers);
                        ch.consume(q2.queue, function() {}, {noAck:false}, function() {
                            consumers ++; if (consumers % 1000 == 0) console.log('%d consumers', consumers);
                            setTimeout(queuesDone, 10000);
                        });
                    });
                });
            });
        });
    }
}