<commit_before>var http = require('http');

function buffer_concat(buffers,nread){
    var buffer = null;
    switch(buffers.length) {
        case 0: buffer = new Buffer(0);
            break;
        case 1: buffer = buffers[0];
            break;
        default:
            buffer = new Buffer(nread);
            for (var i = 0, pos = 0, l = buffers.length; i < l; i++) {
                var chunk = buffers[i];
                chunk.copy(buffer, pos);
                pos += chunk.length;
            }
        break;
    }
    return buffer.toString()
}

function errorHandler(err, message) {
    if (err) {
        if (err.socket && (err.statusCode < 200 || err.statusCode > 299)) {
            var data = '';
            var buffers = [];
            var nread = 0;
            err.on('data', function (chunk) {
                buffers.push(chunk);
                nread += chunk.length;
            });
            err.on('end', function () {
                data = buffer_concat(buffers, nread);
                console.log("Host %s responded to %s request on endpoint %s with an error: \n %s".red,
                    err.socket._httpMessage._headers.host, err.socket._httpMessage.method, err.socket._httpMessage.path, data);
            });
        }
        if (err.message) {
            console.log(err.message.red);
        }
        if (message) {
            console.log(message.red);
        }
    }
}

function httpOptions(httpProxy, host, port, auth, path) {
    return httpOptions(httpProxy, host, port, auth, path, 'GET', {});
}

function httpOptions(httpProxy, host, port, auth, path, method, headers) {
    if(httpProxy) {
        var httpProxySplit = httpProxy.split(":");
        var fullPath = 'http://' + host + ':' + port + path
        return { host: httpProxySplit[0], port: httpProxySplit[1], path: fullPath, auth: auth, method: method, headers: headers };
    } else {
        return { host: host, port: port, path: path, auth: auth };
    }
}

function sourceGetHttpOptions(opts, path) {
    return httpOptions(opts.httpProxy, opts.sourceHost, opts.sourcePort, opts.sourceAuth, path);
}

function sourcePostHttpOptions(opts, path, headers) {
    return httpOptions(opts.httpProxy, opts.sourceHost, opts.sourcePort, opts.SourceAuth, path, 'POST', headers);
}

function targetPutHttpOptions(opts, path, headers) {
    return httpOptions(opts.httpProxy, opts.targetHost, opts.targetPort, opts.targetAuth, path, 'PUT', headers);
}

function targetPostHttpOptions(opts, path, headers) {
    return httpOptions(opts.httpProxy, opts.targetHost, opts.targetPort, opts.targetAuth, path, 'POST', headers);
}

/**
 * Resets all stored states of this driver and allows to start over from the beginning without restarting.
 */
exports.reset = function(opts) {
    http.globalAgent.maxSockets = opts.maxSockets;
    exports.scrollId = null;
};

/**
 * Fetches general statistical in informational data from the target database.
 * @param opts
 * @param callback Callback function without a parameter. The stats result will be attached to the opts object.
 */
exports.getTargetStats = function(opts, callback) {
    var tmpOpts = {
        sourceHost: opts.targetHost,
        sourcePort: opts.targetPort,
        sourceAuth: opts.targetAuth,
        httpProxy: opts.httpProxy
    };
    exports.getSourceStats(tmpOpts, function() {
        opts.targetStats = tmpOpts.sourceStats;
        callback();
    });
};

/**
 * Fetches general statistical in informational data from the source database.
 * @param opts
 * @param callback Callback function without a parameter. The stats result will be attached to the opts object.
 */
exports.getSourceStats = function(opts, callback) {
    if (opts.logEnabled) {
        console.log('HTTP Proxy: ' + opts.httpProxy);
        console.log('Reading source statistics from ElasticSearch');
    }

    opts.sourceStats = {
        version: false,
        cluster_status: false,
        docs: false,
        aliases: false
    };
    function done() {
        for (var prop in opts.sourceStats) {
            if (!opts.sourceStats[prop]) {
                return;
            }
        }
        opts.sourceStats['retries'] = 0;
        callback();<commit_msg>added proper parsing of http proxy url<commit_after>var http = require('http');
var url = require('url');

function buffer_concat(buffers,nread){
    var buffer = null;
    switch(buffers.length) {
        case 0: buffer = new Buffer(0);
            break;
        case 1: buffer = buffers[0];
            break;
        default:
            buffer = new Buffer(nread);
            for (var i = 0, pos = 0, l = buffers.length; i < l; i++) {
                var chunk = buffers[i];
                chunk.copy(buffer, pos);
                pos += chunk.length;
            }
        break;
    }
    return buffer.toString()
}

function errorHandler(err, message) {
    if (err) {
        if (err.socket && (err.statusCode < 200 || err.statusCode > 299)) {
            var data = '';
            var buffers = [];
            var nread = 0;
            err.on('data', function (chunk) {
                buffers.push(chunk);
                nread += chunk.length;
            });
            err.on('end', function () {
                data = buffer_concat(buffers, nread);
                console.log("Host %s responded to %s request on endpoint %s with an error: \n %s".red,
                    err.socket._httpMessage._headers.host, err.socket._httpMessage.method, err.socket._httpMessage.path, data);
            });
        }
        if (err.message) {
            console.log(err.message.red);
        }
        if (message) {
            console.log(message.red);
        }
    }
}

function httpOptions(httpProxy, host, port, auth, path, method, headers) {
    if(httpProxy) {
        var httpProxyUrl = url.parse(httpProxy)
        var fullPath = 'http://' + host + ':' + port + path
        headers["Host"] = httpProxy
        return { host: httpProxyUrl.hostname, port: httpProxyUrl.port, path: fullPath, auth: auth, method: method, headers: headers };
    } else {
        return { host: host, port: port, path: path, auth: auth };
    }
}

function sourceGetHttpOptions(opts, path) {
    return httpOptions(opts.httpProxy, opts.sourceHost, opts.sourcePort, opts.sourceAuth, path, 'GET', {});
}

function sourcePostHttpOptions(opts, path, headers) {
    return httpOptions(opts.httpProxy, opts.sourceHost, opts.sourcePort, opts.SourceAuth, path, 'POST', headers);
}

function targetPutHttpOptions(opts, path, headers) {
    return httpOptions(opts.httpProxy, opts.targetHost, opts.targetPort, opts.targetAuth, path, 'PUT', headers);
}

function targetPostHttpOptions(opts, path, headers) {
    return httpOptions(opts.httpProxy, opts.targetHost, opts.targetPort, opts.targetAuth, path, 'POST', headers);
}

/**
 * Resets all stored states of this driver and allows to start over from the beginning without restarting.
 */
exports.reset = function(opts) {
    http.globalAgent.maxSockets = opts.maxSockets;
    exports.scrollId = null;
};

/**
 * Fetches general statistical in informational data from the target database.
 * @param opts
 * @param callback Callback function without a parameter. The stats result will be attached to the opts object.
 */
exports.getTargetStats = function(opts, callback) {
    var tmpOpts = {
        sourceHost: opts.targetHost,
        sourcePort: opts.targetPort,
        sourceAuth: opts.targetAuth,
        httpProxy: opts.httpProxy
    };
    exports.getSourceStats(tmpOpts, function() {
        opts.targetStats = tmpOpts.sourceStats;
        callback();
    });
};

/**
 * Fetches general statistical in informational data from the source database.
 * @param opts
 * @param callback Callback function without a parameter. The stats result will be attached to the opts object.
 */
exports.getSourceStats = function(opts, callback) {
    if (opts.logEnabled) {
        console.log('Reading source statistics from ElasticSearch');
    }

    opts.sourceStats = {
        version: false,
        cluster_status: false,
        docs: false,
        aliases: false
    };
    function done() {
        for (var prop in opts.sourceStats) {
            if (!opts.sourceStats[prop]) {
                return;
            }
        }
        opts.sourceStats['retries'] = 0;
        callback();
{'fraction_non_alphanumeric': 0.11817528735632184, 'fraction_numerical': 0.0035919540229885057, 'mean_word_length': 2.412173202614379, 'pattern_counts': {'":': 1, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1144539', 'n_tokens_mistral': 2377, 'n_tokens_neox': 2332, 'n_words': 657}