<commit_before>var elasticsearch = require('elasticsearch'),
    Generator = require('./mapping-generator'),
    generator = new Generator(),
    serialize = require('./serialize'),
    events = require('events'),
    nop = require('nop'),
    util = require('util');

var GlobalDuringIndexCreationQueue = {};

module.exports = function Mongoosastic(schema, options) {
    options = options || {};

    var bulkTimeout, bulkBuffer = [], esClient,
        mapping = getMapping(schema),
        indexName = options && options.index,
        typeName = options && options.type,
        alwaysHydrate = options && options.hydrate,
        defaultHydrateOptions = options && options.hydrateOptions,
        bulk = options && options.bulk,
        filter = options && options.filter,
        bulkCallbacks = [],
        routePath = options.routePath;

    if (options.esClient) {
        esClient = options.esClient;
    } else {
        esClient = createEsClient(options);
    }

    if (typeof routePath === 'string') {
        routePath = routePath;
    } else {
        routePath = null;
    }

    setUpMiddlewareHooks(schema);

    /**
     * ElasticSearch Client
     */
    schema.statics.esClient = esClient;
    schema.statics.esIndexName = indexName;
    schema.statics.esTypeName = typeName;

    /**
     * Create the mapping. Takes an optional settings parameter and a callback that will be called once
     * the mapping is created

     * @param settings Object (optional)
     * @param cb Function
     */
    schema.statics.createMapping = function (settings, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            settings = undefined;
        }

        setIndexNameIfUnset(this.modelName);

        createMappingIfNotPresent({
            client: esClient,
            indexName: indexName,
            typeName: typeName,
            schema: schema,
            settings: settings
        }, cb);
    };

    /**
     * Create the EXPLICIT mapping. Takes an optional indexSettings parameter and a callback that will be called once
     * the mapping is created
     * @param explicitMapping Object
     * @param indexSettings Object (optional)
     * @param cb Function
     */
    schema.statics.createExplicitMapping = function (explicitMapping, indexSettings, cb) {
        if (arguments.length < 3) {
            cb = arguments[1] || nop;
            indexSettings = undefined;
        }
        setIndexNameIfUnset(this.modelName);
        createMappingIfNotPresent({
            client: esClient,
            indexName: indexName,
            typeName: typeName,
            explicitMapping: explicitMapping,
            schema: schema,
            settings: indexSettings
        }, cb);
    };

    /**
     * @param options  Object (optional)
     * @param cb Function
     */
    schema.methods.index = function (options, cb) {
        var _cb = null;
        if (arguments.length < 2) {
            _cb = arguments[0];
            options = {};
        } else {
            _cb = cb;
        }

        if (filter && filter(this) && typeof _cb === 'function') {
            return _cb();
        }

        setIndexNameIfUnset(this.constructor.modelName);

        var index = options.index || indexName,
            type = options.type || typeName,
            routeValue = null,
            finalParam = null;

        if (routePath) {
            routeValue = this.get(routePath);
            if (routeValue === undefined || routeValue === null) {
                routeValue = this.get('_id');
            }
            if (typeof routeValue !== 'string') {
                routeValue = routeValue.toString();
            }
        }
        var serialModel = serialize(this, mapping);
        if (bulk) {
            /**
             * To serialize in bulk it needs the _id
             */
            serialModel._id = this._id;
            finalParam = {
                index: index,
                type: type,
                model: serialModel
            };
            if (routeValue) {
                finalParam['routing'] = routeValue;
            }
            bulkIndex(finalParam, _cb);
        } else {
            finalParam = {
                index: index,
                type: type,
                id: this._id.toString(),
                body: {
                    doc: serialModel,
                    upsert: serialModel
                },
                retryOnConflict: 3
            };
            if (routeValue) {
                finalParam['routing'] = routeValue;
            }
            esClient.update(finalParam, _cb);
        }
    };

    /**
     * Unset elasticsearch index
     * @param options - (optional) options for unIndex
     * @param cb - callback when unIndex is complete
     */
    schema.methods.unIndex = function (options, cb) {
        var _cb = null;
        if (arguments.length < 2) {
            _cb = arguments[0] || nop;
            options = {};
        } else {
            _cb = cb;
        }

        setIndexNameIfUnset(this.constructor.modelName);

        options.index = options.index || indexName;
        options.type = options.type || typeName;
        options.model = this;
        options.client = esClient;
        options.tries = 3;

        if (bulk)
            bulkDelete(options, _cb);
        else
            deleteByMongoId(options, _cb);
    };

    /**
     * Delete all documents from a type/index
     * @param options - (optional) specify index/type
     * @param cb - callback when truncation is complete
     */
    schema.statics.esTruncate = function (options, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            options = {};
        }

        setIndexNameIfUnset(this.modelName);

        var index = options.index || indexName,
            type = options.type || typeName;

        esClient.deleteByQuery({
            index: index,
            type: type,
            body: {
                query: {
                    match_all: {}
                }
            }
        }, cb);
    };

    /**
     * Synchronize an existing collection
     *
     * @param query - query for documents you want to synchronize
     */
    schema.statics.synchronize = function (query) {
        var em = new events.EventEmitter(),
            closeValues = [],
            pending = 0,
            close = function () {
                em.emit.apply(em, ['close'].concat(closeValues));
            };

        //Set indexing to be bulk when synchronizing to make synchronizing faster
        bulk = {
                delay: 1000,
                size: 1000
            };

        query = query || {};

        setIndexNameIfUnset(this.modelName);

        var stream = this.find(query).stream();

        stream.on('data', function (doc) {
            pending++;
            if (pending > 5 * bulk.size && !stream.paused) {
                stream.pause();
            } else if (pending <= bulk.size && stream.paused) {
                stream.resume();
            }
            doc.index(function (err, doc) {
                pending--;
                if (err) {
                    em.emit('error', err);
                } else {
                    em.emit('data', null, doc);
                }
            });
        });

        stream.on('close', function (a, b) {
            closeValues = [a, b];
            var closeInterval = setInterval(function () {
                if (pending === 0 && bulkBuffer.length === 0) {
                    clearInterval(closeInterval);
                    close();
                }
            }, 100);
        });

        stream.on('error', function (err) {
            em.emit('error', err);
        });

        return em;
    };
    /**
     * ElasticSearch search function
     *
     * @param query - query object to perform search with
     * @param options - (optional) special search options, such as hydrate
     * @param cb - callback called with search results
     */
    schema.statics.search = function (query, options, cb) {
        var _cb = null;
        if (arguments.length === 2) {
            _cb = arguments[1];
            options = {};
        } else {
            _cb = cb;
        }

        options.hydrateOptions = options.hydrateOptions || defaultHydrateOptions || {};

        if (query === null)
            query = undefined;

        var _this = this,
            esQuery = {
                body: query,
                index: options.index || indexName,
                type: options.type || typeName
            };
        if (options.highlight) {
            esQuery.body.highlight = options.highlight;
            delete options.highlight;
        }

        Object.keys(options).forEach(function (opt) {
            if (!opt.match(/hydrate/) && options.hasOwnProperty(opt))
                esQuery[opt] = options[opt];
        });

        setIndexNameIfUnset(this.modelName);

        esClient.search(esQuery, function (err, res) {
            if (err) {
                return cb(err);
            }

            if (alwaysHydrate || options.hydrate) {
                hydrate(res, _this, options, cb);
            } else {
                cb(null, res);
            }
        });
    };

    schema.statics.esCount = function (query, cb) {
        var _cb = null;
        setIndexNameIfUnset(this.modelName);

        if (cb == null && typeof query === 'function') {
            _cb = query;
            query = null;
        } else {
            _cb = cb;
        }

        var esQuery = {
            body: {
                query: query
            },
            index: options.index || indexName,
            type: options.type || typeName
        };

        esClient.count(esQuery, _cb);
    };

    /**
     * Create the mapping. Takes an optional settings parameter and a callback that will be called once
     * the mapping is created

     * @param settings Object (optional)
     * @param cb Function
     */
    schema.statics.createMapping = function (settings, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            settings = undefined;
        }

        setIndexNameIfUnset(this.modelName);

        createMappingIfNotPresent({
            client: esClient,
            indexName: indexName,
            typeName: typeName,
            schema: schema,
            settings: settings
        }, cb);
    };

    function bulkDelete(options, cb) {
        bulkAdd({
            delete: {
                _index: options.index || indexName,
                _type: options.type || typeName,
                _id: options.model._id.toString()
            }
        },null, cb);
    }

    function bulkIndex(options, cb) {
        var finalParam = {
            update: {
                _index: options.index || indexName,
                _type: options.type || typeName,
                _id: options.model._id.toString(),
                _retry_on_conflict: 3
            }
        };
        if (options.routing) {
            finalParam.update['_routing'] = options.routing;
        }
        bulkAdd(finalParam, {doc:options.model, upsert: options.model}, cb);
    }

    function clearBulkTimeout() {
        clearTimeout(bulkTimeout);
        bulkTimeout = undefined;
    }

    function bulkAdd(instruction, data, cb) {
        bulkBuffer.push(instruction);
        bulkCallbacks.push(cb);
        if (data) {
            bulkBuffer.push(data);
        }

        if (bulkBuffer.length >= (bulk.size || 1000)) {
            schema.statics.flush();
            clearBulkTimeout();
        } else if (bulkTimeout === undefined) {
            bulkTimeout = setTimeout(
            function () {
                schema.statics.flush();
                clearBulkTimeout();
            }, bulk.delay || 1000);
        }
    }

    schema.statics.flush = function (cb) {
        var _cb = cb || nop;

        var thisBulkBuffer = bulkBuffer;
        var thisBulkCallbacks = bulkCallbacks;

        bulkBuffer = [];
        bulkCallbacks = [];

        esClient.bulk({
            body: thisBulkBuffer
        }, function (err, results) {
            var _err = null;

            if (!err && results && results.items && results.items.length === thisBulkCallbacks.length) {
                for (var i=results.items.length-1; i >=0; i--) {
                    if (typeof thisBulkCallbacks[i] === 'function') {
                        thisBulkCallbacks[i](null, results.items[i]);
                    }

                }
            } else {

                _err = err || new Error('Bulk result count dismatched query count.');
                for (var j=thisBulkCallbacks.length- 1; j>=0; j--) {
                    if (typeof thisBulkCallbacks[j] === 'function') {
                        thisBulkCallbacks[j](_err);
                    }
                }

            }
            thisBulkBuffer = null;
            thisBulkCallbacks = null;
            return _cb(_err, results);

        });

    };

    schema.statics.refresh = function (options, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            options = {};
        }

        setIndexNameIfUnset(this.modelName);
        esClient.indices.refresh({
            index: options.index || indexName<commit_msg>Try fix 'memory leak'<commit_after>var elasticsearch = require('elasticsearch'),
    Generator = require('./mapping-generator'),
    generator = new Generator(),
    serialize = require('./serialize'),
    events = require('events'),
    nop = require('nop'),
    util = require('util');

var GlobalDuringIndexCreationQueue = {};

module.exports = function Mongoosastic(schema, options) {
    options = options || {};

    var bulkTimeout, bulkBuffer = [], esClient,
        mapping = getMapping(schema),
        indexName = options && options.index,
        typeName = options && options.type,
        alwaysHydrate = options && options.hydrate,
        defaultHydrateOptions = options && options.hydrateOptions,
        bulk = options && options.bulk,
        filter = options && options.filter,
        bulkCallbacks = [],
        routePath = options.routePath;

    if (options.esClient) {
        esClient = options.esClient;
    } else {
        esClient = createEsClient(options);
    }

    if (typeof routePath === 'string') {
        routePath = routePath;
    } else {
        routePath = null;
    }

    setUpMiddlewareHooks(schema);

    /**
     * ElasticSearch Client
     */
    schema.statics.esClient = esClient;
    schema.statics.esIndexName = indexName;
    schema.statics.esTypeName = typeName;

    /**
     * Create the mapping. Takes an optional settings parameter and a callback that will be called once
     * the mapping is created

     * @param settings Object (optional)
     * @param cb Function
     */
    schema.statics.createMapping = function (settings, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            settings = undefined;
        }

        setIndexNameIfUnset(this.modelName);

        createMappingIfNotPresent({
            client: esClient,
            indexName: indexName,
            typeName: typeName,
            schema: schema,
            settings: settings
        }, cb);
    };

    /**
     * Create the EXPLICIT mapping. Takes an optional indexSettings parameter and a callback that will be called once
     * the mapping is created
     * @param explicitMapping Object
     * @param indexSettings Object (optional)
     * @param cb Function
     */
    schema.statics.createExplicitMapping = function (explicitMapping, indexSettings, cb) {
        if (arguments.length < 3) {
            cb = arguments[1] || nop;
            indexSettings = undefined;
        }
        setIndexNameIfUnset(this.modelName);
        createMappingIfNotPresent({
            client: esClient,
            indexName: indexName,
            typeName: typeName,
            explicitMapping: explicitMapping,
            schema: schema,
            settings: indexSettings
        }, cb);
    };

    /**
     * @param options  Object (optional)
     * @param cb Function
     */
    schema.methods.index = function (options, cb) {
        var _cb = null;
        if (arguments.length < 2) {
            _cb = arguments[0];
            options = {};
        } else {
            _cb = cb;
        }

        if (filter && filter(this) && typeof _cb === 'function') {
            return _cb();
        }

        setIndexNameIfUnset(this.constructor.modelName);

        var index = options.index || indexName,
            type = options.type || typeName,
            routeValue = null,
            finalParam = null;

        if (routePath) {
            routeValue = this.get(routePath);
            if (routeValue === undefined || routeValue === null) {
                routeValue = this.get('_id');
            }
            if (typeof routeValue !== 'string') {
                routeValue = routeValue.toString();
            }
        }
        var serialModel = serialize(this, mapping);
        if (bulk) {
            /**
             * To serialize in bulk it needs the _id
             */
            serialModel._id = this._id;
            finalParam = {
                index: index,
                type: type,
                model: serialModel
            };
            if (routeValue) {
                finalParam['routing'] = routeValue;
            }
            bulkIndex(finalParam, _cb);
        } else {
            finalParam = {
                index: index,
                type: type,
                id: this._id.toString(),
                body: {
                    doc: serialModel,
                    upsert: serialModel
                },
                retryOnConflict: 3
            };
            if (routeValue) {
                finalParam['routing'] = routeValue;
            }
            esClient.update(finalParam, _cb);
        }
    };

    /**
     * Unset elasticsearch index
     * @param options - (optional) options for unIndex
     * @param cb - callback when unIndex is complete
     */
    schema.methods.unIndex = function (options, cb) {
        var _cb = null;
        if (arguments.length < 2) {
            _cb = arguments[0] || nop;
            options = {};
        } else {
            _cb = cb;
        }

        setIndexNameIfUnset(this.constructor.modelName);

        options.index = options.index || indexName;
        options.type = options.type || typeName;
        options.model = this;
        options.client = esClient;
        options.tries = 3;

        if (bulk)
            bulkDelete(options, _cb);
        else
            deleteByMongoId(options, _cb);
    };

    /**
     * Delete all documents from a type/index
     * @param options - (optional) specify index/type
     * @param cb - callback when truncation is complete
     */
    schema.statics.esTruncate = function (options, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            options = {};
        }

        setIndexNameIfUnset(this.modelName);

        var index = options.index || indexName,
            type = options.type || typeName;

        esClient.deleteByQuery({
            index: index,
            type: type,
            body: {
                query: {
                    match_all: {}
                }
            }
        }, cb);
    };

    /**
     * Synchronize an existing collection
     *
     * @param query - query for documents you want to synchronize
     */
    schema.statics.synchronize = function (query) {
        var em = new events.EventEmitter(),
            closeValues = [],
            pending = 0,
            close = function () {
                em.emit.apply(em, ['close'].concat(closeValues));
            };

        //Set indexing to be bulk when synchronizing to make synchronizing faster
        bulk = {
                delay: 1000,
                size: 1000
            };

        query = query || {};

        setIndexNameIfUnset(this.modelName);

        var stream = this.find(query).stream();

        stream.on('data', function (doc) {
            pending++;
            if (pending > 5 * bulk.size && !stream.paused) {
                stream.pause();
            } else if (pending <= bulk.size && stream.paused) {
                stream.resume();
            }
            doc.index(function (err, doc) {
                pending--;
                if (err) {
                    em.emit('error', err);
                } else {
                    em.emit('data', null, doc);
                }
            });
        });

        stream.on('close', function (a, b) {
            closeValues = [a, b];
            var closeInterval = setInterval(function () {
                if (pending === 0 && bulkBuffer.length === 0) {
                    clearInterval(closeInterval);
                    close();
                }
            }, 100);
        });

        stream.on('error', function (err) {
            em.emit('error', err);
        });

        return em;
    };
    /**
     * ElasticSearch search function
     *
     * @param query - query object to perform search with
     * @param options - (optional) special search options, such as hydrate
     * @param cb - callback called with search results
     */
    schema.statics.search = function (query, options, cb) {
        var _cb = null;
        if (arguments.length === 2) {
            _cb = arguments[1];
            options = {};
        } else {
            _cb = cb;
        }

        options.hydrateOptions = options.hydrateOptions || defaultHydrateOptions || {};

        if (query === null)
            query = undefined;

        var _this = this,
            esQuery = {
                body: query,
                index: options.index || indexName,
                type: options.type || typeName
            };
        if (options.highlight) {
            esQuery.body.highlight = options.highlight;
            delete options.highlight;
        }

        Object.keys(options).forEach(function (opt) {
            if (!opt.match(/hydrate/) && options.hasOwnProperty(opt))
                esQuery[opt] = options[opt];
        });

        setIndexNameIfUnset(this.modelName);

        esClient.search(esQuery, function (err, res) {
            if (err) {
                return cb(err);
            }

            if (alwaysHydrate || options.hydrate) {
                hydrate(res, _this, options, cb);
            } else {
                cb(null, res);
            }
        });
    };

    schema.statics.esCount = function (query, cb) {
        var _cb = null;
        setIndexNameIfUnset(this.modelName);

        if (cb == null && typeof query === 'function') {
            _cb = query;
            query = null;
        } else {
            _cb = cb;
        }

        var esQuery = {
            body: {
                query: query
            },
            index: options.index || indexName,
            type: options.type || typeName
        };

        esClient.count(esQuery, _cb);
    };

    /**
     * Create the mapping. Takes an optional settings parameter and a callback that will be called once
     * the mapping is created

     * @param settings Object (optional)
     * @param cb Function
     */
    schema.statics.createMapping = function (settings, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            settings = undefined;
        }

        setIndexNameIfUnset(this.modelName);

        createMappingIfNotPresent({
            client: esClient,
            indexName: indexName,
            typeName: typeName,
            schema: schema,
            settings: settings
        }, cb);
    };

    function bulkDelete(options, cb) {
        bulkAdd({
            delete: {
                _index: options.index || indexName,
                _type: options.type || typeName,
                _id: options.model._id.toString()
            }
        },null, cb);
    }

    function bulkIndex(options, cb) {
        var finalParam = {
            update: {
                _index: options.index || indexName,
                _type: options.type || typeName,
                _id: options.model._id.toString(),
                _retry_on_conflict: 3
            }
        };
        if (options.routing) {
            finalParam.update['_routing'] = options.routing;
        }
        bulkAdd(finalParam, {doc:options.model, upsert: options.model}, cb);
    }

    function clearBulkTimeout() {
        clearTimeout(bulkTimeout);
        bulkTimeout = undefined;
    }

    function bulkAdd(instruction, data, cb) {
        bulkBuffer.push(instruction);
        bulkCallbacks.push(cb);
        if (data) {
            bulkBuffer.push(data);
        }

        if (bulkBuffer.length >= (bulk.size || 1000)) {
            schema.statics.flush();
            clearBulkTimeout();
        } else if (bulkTimeout === undefined) {
            bulkTimeout = setTimeout(
            function () {
                schema.statics.flush();
                clearBulkTimeout();
            }, bulk.delay || 1000);
        }
    }

    schema.statics.flush = function (cb) {
        var _cb = cb || nop;

        var thisBulkBuffer = _.clone(bulkBuffer);
        var thisBulkCallbacks = _.clone(bulkCallbacks);

        bulkBuffer.length = 0;
        bulkCallbacks.length = 0;

        esClient.bulk({
            body: thisBulkBuffer
        }, function (err, results) {
            var _err = null;

            if (!err && results && results.items && results.items.length === thisBulkCallbacks.length) {
                for (var i=results.items.length-1; i >=0; i--) {
                    if (typeof thisBulkCallbacks[i] === 'function') {
                        thisBulkCallbacks[i](null, results.items[i]);
                    }

                }
            } else {

                _err = err || new Error('Bulk result count dismatched query count.');
                for (var j=thisBulkCallbacks.length- 1; j>=0; j--) {
                    if (typeof thisBulkCallbacks[j] === 'function') {
                        thisBulkCallbacks[j](_err);
                    }
                }

            }
            thisBulkBuffer.length = 0;
            thisBulkCallbacks.length = 0;
            return _cb(_err, results);

        });

    };

    schema.statics.refresh = function (options, cb) {
        if (arguments.length < 2) {
            cb = arguments[0] || nop;
            options = {};
        }

        setIndexNameIfUnset(this.modelName);
        esClient.indices.refresh({
            index: options.index || indexName
{'fraction_non_alphanumeric': 0.11019675319817461, 'fraction_numerical': 0.003590932894441535, 'mean_word_length': 1.4759214669383218, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 11, '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': '5068401', 'n_tokens_mistral': 7252, 'n_tokens_neox': 6931, 'n_words': 1678}