<commit_before>var solrClient = require('solr-client'),
    zkClient = require('node-zookeeper-client'),
    restClient = require('node-rest-client'),
    async = require('async'),
    url = require('url'),
    _ = require('underscore');

var DEFAULT_OPTIONS = {
    zkConnectionString: 'localhost:2181',
    zkLiveNodes: '/live_nodes',
    solrProtocol: 'http',
    solrCollectionsGetEndPoint: '/admin/collections?action=LIST',
    ssh: {},
    // Passed verbatim to node-zookeeper-client
    zk: {
        sessionTimeout: 3000,
        spinDelay : 1000,
        retries : 1
    },
    // Passed verbatim to node-rest-client
    rest: {
        requestConfig: {
            timeout: 3000
        },
        responseConfig: {
            timeout: 3000
        }
    }
};

function createClient(solrCollection, options, callback) {
    options = _.defaults(_.clone(options), DEFAULT_OPTIONS);
    parseSshTunnels(options.ssh);

    async.waterfall([
        function (callback) {
            getZkLiveNodes(options, callback);
        },
        function (solrInstances, callback) {
            getSolrCollections(solrInstances, options, callback);
        },
        function (solrBundles, callback) {
            getSolrInstance(solrBundles, solrCollection, callback);
        },
        function (solrInstance, callback) {
            getSolrParams(solrInstance, options, callback);
        }
    ], function (err, solrParams) {
        if (err) {
            return callback(err);
        }
        callback(null, solrClient.createClient(solrParams.host, solrParams.port, solrParams.core, solrParams.path + '/' + solrCollection));
    });
}

function getSolrParams(solrInstance, options, callback) {
    solrInstance = tunnel(solrInstance, options.ssh.tunnels);
    var urlParams = url.parse(solrInstance);
    callback(null, {
        host: urlParams.hostname,
        port: urlParams.port,
        core: '',
        path: urlParams.path
    })
}

function getSolrInstance(solrBundles, solrCollection, callback) {
    async.filter(solrBundles, function (bundle, callback) {
        async.some(bundle.collections, function (collection, callback) {
            callback(collection == solrCollection);
        }, callback)
    }, function (filteredBundles) {
        async.map(filteredBundles, function (bundle, callback) {
            callback(null, bundle.instance);
        }, function (err, instances) {
            if (err) {
                return callback(err);
            }
            if (!instances.length) {
                return callback('Found no Solr instance hosting collection \'' + solrCollection + '\'');
            }
            var instance = instances[Math.floor(Math.random() * instances.length)];
            callback(null, instance);
        })
    });
}

function getSolrCollections(solrInstances, options, callback) {
    var client = new restClient.Client();

    var tasks = solrInstances.map(function (instance) {
        if (options.ssh && options.ssh.tunnels) {
            instance = tunnel(instance, options.ssh.tunnels);
        }
        var protocolInstance = options.solrProtocol + '://' + instance;
        return function (callback) {
            client.get(protocolInstance + options.solrCollectionsGetEndPoint, options.rest, function (data /*, response*/) {
                var bundle = {
                    instance: protocolInstance,
                    collections: data.response.arr[0].str
                };
                callback(null, bundle);
            });
        };
    });

    async.parallel(tasks, callback);
}

function getZkLiveNodes(options, callback) {
    var client = null;

    function list(callback) {
        client.getChildren(options.zkLiveNodes, null, function (err, children) {
            if (err) {
                return callback(err);
            }
            if (children.length == 0) {
                callback('Found no live Solr nodes under path \'' + options.zkLiveNodes + '\' by connecting at \'' + options.zkConnectionString + '\'');
            }
            async.map(children, function (item, callback) {
                var url = replaceAfterPort(item, /_/g, '/');
                callback(null, url);
            }, function (err, mappedChildren) {
                client.close();
                callback(err, mappedChildren);
            });
        });
    }<commit_msg>changed get collections list response type to json and fixed minor bug with filtering instances<commit_after>var solrClient = require('solr-client'),
    zkClient = require('node-zookeeper-client'),
    restClient = require('node-rest-client'),
    async = require('async'),
    url = require('url'),
    _ = require('underscore');

var DEFAULT_OPTIONS = {
    zkConnectionString: 'localhost:2181',
    zkLiveNodes: '/live_nodes',
    solrProtocol: 'http',
    solrCollectionsGetEndPoint: '/admin/collections?action=LIST&wt=json',
    ssh: {},
    // Passed verbatim to node-zookeeper-client
    zk: {
        sessionTimeout: 3000,
        spinDelay : 1000,
        retries : 1
    },
    // Passed verbatim to node-rest-client
    rest: {
        requestConfig: {
            timeout: 3000
        },
        responseConfig: {
            timeout: 3000
        }
    }
};

function createClient(solrCollection, options, callback) {
    options = _.defaults(_.clone(options), DEFAULT_OPTIONS);
    parseSshTunnels(options.ssh);

    async.waterfall([
        function (callback) {
            getZkLiveNodes(options, callback);
        },
        function (solrInstances, callback) {
            getSolrCollections(solrInstances, options, callback);
        },
        function (solrBundles, callback) {
            getSolrInstance(solrBundles, solrCollection, callback);
        },
        function (solrInstance, callback) {
            getSolrParams(solrInstance, options, callback);
        }
    ], function (err, solrParams) {
        if (err) {
            return callback(err);
        }
        callback(null, solrClient.createClient(solrParams.host, solrParams.port, solrParams.core, solrParams.path + '/' + solrCollection));
    });
}

function getSolrParams(solrInstance, options, callback) {
    solrInstance = tunnel(solrInstance, options.ssh.tunnels);
    var urlParams = url.parse(solrInstance);
    callback(null, {
        host: urlParams.hostname,
        port: urlParams.port,
        core: '',
        path: urlParams.path
    })
}

function getSolrInstance(solrBundles, solrCollection, callback) {
    async.filter(solrBundles, function (bundle, callback) {
        callback(bundle.collections == solrCollection);
    }, function (filteredBundles) {
        async.map(filteredBundles, function (bundle, callback) {
            callback(null, bundle.instance);
        }, function (err, instances) {
            if (err) {
                return callback(err);
            }
            if (!instances.length) {
                return callback('Found no Solr instance hosting collection \'' + solrCollection + '\'');
            }
            var instance = instances[Math.floor(Math.random() * instances.length)];
            callback(null, instance);
        })
    });
}

function getSolrCollections(solrInstances, options, callback) {
    var client = new restClient.Client();

    var tasks = solrInstances.map(function (instance) {
        if (options.ssh && options.ssh.tunnels) {
            instance = tunnel(instance, options.ssh.tunnels);
        }
        var protocolInstance = options.solrProtocol + '://' + instance;
        return function (callback) {
            client.get(protocolInstance + options.solrCollectionsGetEndPoint, options.rest, function (data /*, response*/) {
                data = JSON.parse(data);
                var bundle = {
                    instance: protocolInstance,
                    collections: data.collections[0]
                };
                callback(null, bundle);
            });
        };
    });

    async.parallel(tasks, callback);
}

function getZkLiveNodes(options, callback) {
    var client = null;

    function list(callback) {
        client.getChildren(options.zkLiveNodes, null, function (err, children) {
            if (err) {
                return callback(err);
            }
            if (children.length == 0) {
                callback('Found no live Solr nodes under path \'' + options.zkLiveNodes + '\' by connecting at \'' + options.zkConnectionString + '\'');
            }
            async.map(children, function (item, callback) {
                var url = replaceAfterPort(item, /_/g, '/');
                callback(null, url);
            }, function (err, mappedChildren) {
                client.close();
                callback(err, mappedChildren);
            });
        });
    }
{'fraction_non_alphanumeric': 0.11870011402508551, 'fraction_numerical': 0.00524515393386545, 'mean_word_length': 2.1836660617059893, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5847404', 'n_tokens_mistral': 2479, 'n_tokens_neox': 2361, 'n_words': 495}