lib/sync.js
function (next) {
            // if no documents to index, skip population
            if (!docsToIndex) {
                return next()
            }

            // stream docs - and upload in batches of size BATCH_SIZE
            var docStream = self.find().stream()

            // elasticsearch commands to perform batch-indexing
            var commandSequence = [];

            docStream.on('data', function (doc) {
                if (!doc || !doc._id) {
                    excludedDocs++;
                    return
                }

                // get rid of mongoose-added functions
                doc = helpers.serializeModel(doc,options)

                var selfStream = this
                var strObjectId = doc._id

                var command = {
                    index: {
                        _index: versionedIndexName,
                        _type: options.type,
                        _id: strObjectId
                    }
                }

                // append elasticsearch command and JSON-ified doc to command
                commandSequence.push(command)
                commandSequence.push(doc)


                if (commandSequence.length === BATCH_SIZE) {
                    // pause the stream of incoming docs until we're done
                    // indexing the batch in elasticsearch
                    selfStream.pause()

                    exports.bulkIndexRequest(versionedIndexName, commandSequence, options, function (err) {
                        if (err) {
                            return next(err)
                        }

                        // empty our commandSequence
                        commandSequence = []

                        // keep streaming now that we're ready to accept more
                        selfStream.resume()
                    })
                }
            })

            docStream.on('close', function () {
                // if no documents left in buffer, don't perform a bulk index request
                if (!commandSequence.length) {
                    return next()
                }

                // take care of the rest of the docs left in the buffer
                exports.bulkIndexRequest(versionedIndexName, commandSequence, options, function (err) {
                    if (err) {
                        return next(err)
                    }

                    // empty the commandSequence
                    commandSequence = []

                    return next()
                })
            })
        }
function (next) {
            // if no documents to index, skip population
            if (!docsToIndex) {
                return next()
            }

            // stream docs - and upload in batches of size BATCH_SIZE
            var docStream = self.find().stream()

            // elasticsearch commands to perform batch-indexing
            var commandSequence = [];

            docStream.on('data', function (doc) {
                if (!doc || !doc._id) {
                    excludedDocs++;
                    return
                }

                // get rid of mongoose-added functions
                doc = helpers.serializeModel(doc,options)

                var selfStream = this
                var strObjectId = doc._id

                var command = {
                    index: {
                        _index: versionedIndexName,
                        _type: options.type,
                        _id: strObjectId
                    }
                }

                // append elasticsearch command and JSON-ified doc to command
                commandSequence.push(command)
                commandSequence.push(doc)


                if (commandSequence.length === BATCH_SIZE) {
                    // pause the stream of incoming docs until we're done
                    // indexing the batch in elasticsearch
                    selfStream.pause()

                    exports.bulkIndexRequest(versionedIndexName, commandSequence, options, function (err) {
                        if (err) {
                            return next(err)
                        }

                        // empty our commandSequence
                        commandSequence = []

                        // keep streaming now that we're ready to accept more
                        selfStream.resume()
                    })
                }
            })

            docStream.on('close', function () {
                // if no documents left in buffer, don't perform a bulk index request
                if (!commandSequence.length) {
                    return next()
                }

                // take care of the rest of the docs left in the buffer
                exports.bulkIndexRequest(versionedIndexName, commandSequence, options, function (err) {
                    if (err) {
                        return next(err)
                    }

                    // empty the commandSequence
                    commandSequence = []

                    return next()
                })
            })
        }
function ( next ) { // if no documents to index, skip population if ( ! docsToIndex ) { return next ( ) } // stream docs - and upload in batches of size BATCH_SIZE var docStream = self . find ( ) . stream ( ) // elasticsearch commands to perform batch-indexing var commandSequence = [ ] ; docStream . on ( 'data' , function ( doc ) { if ( ! doc || ! doc . _id ) { excludedDocs ++ ; return } // get rid of mongoose-added functions doc = helpers . serializeModel ( doc , options ) var selfStream = this var strObjectId = doc . _id var command = { index : { _index : versionedIndexName , _type : options . type , _id : strObjectId } } // append elasticsearch command and JSON-ified doc to command commandSequence . push ( command ) commandSequence . push ( doc ) if ( commandSequence . length === BATCH_SIZE ) { // pause the stream of incoming docs until we're done // indexing the batch in elasticsearch selfStream . pause ( ) exports . bulkIndexRequest ( versionedIndexName , commandSequence , options , function ( err ) { if ( err ) { return next ( err ) } // empty our commandSequence commandSequence = [ ] // keep streaming now that we're ready to accept more selfStream . resume ( ) } ) } } ) docStream . on ( 'close' , function ( ) { // if no documents left in buffer, don't perform a bulk index request if ( ! commandSequence . length ) { return next ( ) } // take care of the rest of the docs left in the buffer exports . bulkIndexRequest ( versionedIndexName , commandSequence , options , function ( err ) { if ( err ) { return next ( err ) } // empty the commandSequence commandSequence = [ ] return next ( ) } ) } ) }
populate the newly created index with this collection's documents
populate the newly created index with this collection s documents