traxxx/src/utils/http.js

'use strict';

const config = require('config');
const tunnel = require('tunnel');
const bhttp = require('bhttp');
const taskQueue = require('promise-task-queue');

const logger = require('../logger')(__filename);

const proxyAgent = tunnel.httpsOverHttp({
    proxy: {
        host: config.proxy.host,
        port: config.proxy.port,
    },
});

function useProxy(url) {
    if (!config.proxy.enable) {
        return false;
    }

    const { hostname } = new URL(url);
    return config.proxy.hostnames.includes(hostname);
}

const queue = taskQueue();

queue.on('concurrencyReached:httpGet', () => {
    logger.silly('Queueing GET requests');
});

queue.on('concurrencyReached:httpPost', () => {
    logger.silly('Queueing POST requests');
});

queue.define('http', async ({
    url,
    method = 'GET',
    body,
    timeout = 30000,
    options = {},
}) => {
    if (body) {
        logger.silly(`${method.toUpperCase()} ${url} with ${body}`);
    } else {
        logger.silly(`${method.toUpperCase()} ${url}`);
    }

    const reqOptions = {
        responseTimeout: timeout,
        ...options,
    };

    if (useProxy(url)) {
        reqOptions.agent = proxyAgent;
    }

    const res = ['POST', 'PUT', 'PATCH'].includes(method.toUpperCase())
        ? await bhttp[method.toLowerCase()](url, body, reqOptions)
        : await bhttp[method.toLowerCase()](url, reqOptions);

    return {
        ...res,
        code: res.statusCode,
    };
}, {
    concurrency: 20,
});

async function get(url, options) {
    return queue.push('http', {
        method: 'get',
        url,
        options,
    });
}

async function post(url, body, options) {
    return queue.push('http', {
        url,
        body,
        options,
    });
}

module.exports = {
    get,
    post,
};