284-qpidd.js

'use strict';
const _ = require('lodash'),
      Promise = require('bluebird'),
      amqp = require('.');

const config = {
  address: 'amqp://192.168.1.6',
  receiverLinkPrefix: 'test-queue'
}

var errorHandler = (partitionId, err) => console.warn(`==> RX ERROR (${partitionId}): ${err}`);

var messagesCounter = 0;
var messageHandler = (partitionId, msg) => {
  ++messagesCounter;
  if (messagesCounter == 10000) {
    console.timeEnd('benchmark');
    process.exit(0);
  }
};

var createReceiver = (client, partitionId, address) => {
  return client.createReceiver(address, { attach: { source: { distributionMode: 'copy' } } })
    .then(receiver => {
      receiver.on('message', messageHandler.bind(null, partitionId));
      receiver.on('errorReceived', errorHandler.bind(null, partitionId));
    });
}

function singleClientTest() {
  console.log('single client');
  const client = new amqp.Client();
  client.connect(config.address)
    .tap(() => console.time('benchmark'))
    .then(() => [...Array(4)].map((_, idx) => createReceiver(client, idx, config.receiverLinkPrefix + idx)));
}

function multiClientTest() {
  console.log('multiple clients');
  let clients = [];
  Promise
    .map(_.range(4), (_, idx) => {
      let client = new amqp.Client();
      clients.push(client);
      return client.connect(config.address);
    })
    .tap(() => console.time('benchmark'))
    .map((client, idx) => {
      createReceiver(client, idx, config.receiverLinkPrefix + idx);
    });
}

if (process.argv[2] === 'multi') {
  multiClientTest();
} else {
  singleClientTest();
}