MQ Light

main.js

const mqlight = require('mqlight');
const Promise = require('bluebird');

const MQLIGHT_OPTIONS = { service: 'amqp://localhost:5672' }; // 'amqp://vmeberlitz.southcentralus.cloudapp.azure.com:5672';
const TOPIC = "public";

function createService(id) {
    return new Promise((resolve, reject) => {
        const statService = mqlight.createClient(
            Object.assign({ id }, MQLIGHT_OPTIONS),
            err => err ? reject(err) : resolve(statService)
        );
    });
}

let _lastId = 1;
next_id = () => _lastId++;

function generate(num) {
    let i = 0;
    let arr = [];
    while (i < num) {
        arr.push(i++);
    }
    return arr;
}

async function send(concurrency) {
    const sendClient = await createService();
    await Promise.each(generate(100), async () => {
        await Promise.delay(100);
        sendClient.send(TOPIC, {
            client: sendClient.id,
            concurrency,
            id: next_id(),
            time: new Date()
        });
    });
}



(async () => {
    const statService = await createService('stat_service');
    const timesPerConcurrency = {};

    const calculateAvg = (c) => {
        const sum = timesPerConcurrency[c].reduce((a, b) => a + b);
        return sum / timesPerConcurrency[c].length;
    };


    let timeout;
    const resetTimeout = () => {
        timeout && clearTimeout(timeout);
        timeout = setTimeout(() => {
            statService.stop();

            const tale = Object.keys(timesPerConcurrency).map((c) => {
                return { number_of_concurrent_clients: c, avgTime: calculateAvg(c) };
            });

            require('fs').writeFile('./data.json', JSON.stringify(tale), 'utf8', () => process.exit(0));
        }, 10000);
    }
    statService.subscribe(TOPIC);
    statService.on("message", (data, delivery) => {
        const ellapsed = new Date() - new Date(data.time);
        const times = timesPerConcurrency[data.concurrency] = timesPerConcurrency[data.concurrency] || [];
        times.push(ellapsed);
        resetTimeout();
    });
    statService.on("error", err => console.log(err));

    const concurrencyValues = [1,10,100,250];
    await Promise.each(concurrencyValues, async concurrency => {
        console.log(`Running with ${concurrency} clients ...`);
        await Promise.map(generate(concurrency), async () => {
            await send(concurrency);
        }, { concurrency });
        console.log(`\t:${calculateAvg(concurrency)}\n`);
        return;
    });
})();