A variety of stress test for amqplib

stress.js

var PassThrough = require('stream').PassThrough ||
  require('readable-stream/passthrough');

var HWM = 100;

var tasks = new PassThrough({objectMode: true, highWaterMark: HWM});

var conn, ch;
var i = 0;

// Pre-load some 'tasks'

for (var j=HWM; j > 0; j--) tasks.write('go');
var start = process.hrtime();

process.on('SIGINT', function() {
  var total = process.hrtime(start);
  console.log('Ran for %d seconds, sent %d tasks', total[0], i);
  process.exit(0);
});

// Set a consumer going for each 'task' and send it a message.
// When a consumer completes, add another task.
function recycleTasks() {
  var q; while (q = tasks.read()) {
    conn.createChannel().then(function(ch) {
      var ok = ch.assertQueue('', {autoDelete: true,
                                   durable: false,
                                   exclusive: true});
      ok.then(function(qok) {
        ch.consume(qok.queue, function(err, msg) {
          tasks.write('go');
          ch.close();
        });
        ch.sendToQueue(qok.queue, new Buffer('foobar'));
        i++; if (i % 1000 === 0) console.log('Sent %d', i);
      }).then(null, console.warn);
    });
  }
}

require('../').connect().then(function(conn1) {
  conn = conn1;
  conn.createChannel().then(function(ch1) {
    ch = ch1;
    tasks.on('readable', recycleTasks);
    recycleTasks();
  });
});