async reindex(model, concurrent=1000, interval=100) {
    let SD = await this._getSearchDocument(model)
    // the actual configured ES index and collection for native call
    let index = SD.connections[SD.connection[0]].config.index
    let _type = SD.identity

    let count = 0, errors = 0

    let M = await storage.getModel(model)

    try {
      let body = {query: {term: {model}}}
      // delete-by-query apparently does its own retries
      await new Promise((resolve, reject) => {
        SD.native((err, client) => {
          if (err) { reject(err); return }
          client.deleteByQuery({index, body}, (err, response) => {
            if (err) { reject(err); return }
            if (response.errors) { reject(response.errors); return }
            resolve(response)
          })
        })
      })
      this.log.trace('Search reindex documents deleted', model)
    } catch (e) {
      this.log.trace("Search reindex destroy error", model, e.message)
    }

    for (;;) {
      let objs = await M.find().skip(count).limit(concurrent)
      if (objs.length === 0) break
      count += objs.length

      // bulk format is newline-separated action/data JSON strings
      let docs = await Promise.all(objs.map(obj => this._documentToIndex(model, obj)))
      let body = _.flatten(
        docs.map((d) => [{index: {_type, _id: d.id}}, d] )
      ).map(JSON.stringify).join("\n")

      // Although testing with ES 6.5 and _type deprecated since 5.x, validation error if `_type` is not provided
      //   the waterline-elasticsearch connection does set type too on non-native calls

      try {
        await this._retryLimit('Reindex', () => {
          return new Promise((resolve, reject) => {
            SD.native((err, client) => {
              if (err) { reject(err); return }
              client.bulk({index, body}, (err, response) => {
                if (err) { reject(err); return }
                if (response.errors) { reject(response.errors); return }
                resolve(response)
              })
            })
          })
        })
        this.log.trace('Bulk reindex progress', count, 'documents indexed')
      } catch (e) {
        errors += 1
        this.log.trace("Bulk reindex error", e.message)
      }

      if (objs.length < concurrent) break
      await timeout(interval)
    }
    this.log.trace('Bulk reindex complete', count, 'documents indexed', "with", errors, "errors")
  }
/**
   * Reindex all of a model's documents. Different services concurrent request and queue limits are parameters
   * @param  {string} model the model identity
   * @param  {string} concurrent how many docs to concurrently process
   * @param  {string} interval ms to wait between doc chunks
   */
Reindex all of a model's documents. Different services concurrent request and queue limits are parameters
@param  {string} model the model identity
@param  {string} concurrent how many docs to concurrently process
@param  {string} interval ms to wait between doc chunks
Reindex all of a model ' s documents . Different services concurrent request and queue limits are parameters @param { string } model the model identity @param { string } concurrent how many docs to concurrently process @param { string } interval ms to wait between doc chunks
async reindex(model, concurrent=1000, interval=100) {
    let SD = await this._getSearchDocument(model)
    let index = SD.connections[SD.connection[0]].config.index
    let _type = SD.identity
    let count = 0, errors = 0
    let M = await storage.getModel(model)
    try {
      let body = {query: {term: {model}}}
      await new Promise((resolve, reject) => {
        SD.native((err, client) => {
          if (err) { reject(err); return }
          client.deleteByQuery({index, body}, (err, response) => {
            if (err) { reject(err); return }
            if (response.errors) { reject(response.errors); return }
            resolve(response)
          })
        })
      })
      this.log.trace('Search reindex documents deleted', model)
    } catch (e) {
      this.log.trace("Search reindex destroy error", model, e.message)
    }
    for (;;) {
      let objs = await M.find().skip(count).limit(concurrent)
      if (objs.length === 0) break
      count += objs.length
      let docs = await Promise.all(objs.map(obj => this._documentToIndex(model, obj)))
      let body = _.flatten(
        docs.map((d) => [{index: {_type, _id: d.id}}, d] )
      ).map(JSON.stringify).join("\n")
      try {
        await this._retryLimit('Reindex', () => {
          return new Promise((resolve, reject) => {
            SD.native((err, client) => {
              if (err) { reject(err); return }
              client.bulk({index, body}, (err, response) => {
                if (err) { reject(err); return }
                if (response.errors) { reject(response.errors); return }
                resolve(response)
              })
            })
          })
        })
        this.log.trace('Bulk reindex progress', count, 'documents indexed')
      } catch (e) {
        errors += 1
        this.log.trace("Bulk reindex error", e.message)
      }
      if (objs.length < concurrent) break
      await timeout(interval)
    }
    this.log.trace('Bulk reindex complete', count, 'documents indexed', "with", errors, "errors")
  }
async reindex ( model , concurrent = 1000 , interval = 100 ) { let SD = await this . _getSearchDocument ( model ) let index = SD . connections [ SD . connection [ 0 ] ] . config . index let _type = SD . identity let count = 0 , errors = 0 let M = await storage . getModel ( model ) try { let body = { query : { term : { model } } } await new Promise ( ( resolve , reject ) => { SD . native ( ( err , client ) => { if ( err ) { reject ( err ) ; return } client . deleteByQuery ( { index , body } , ( err , response ) => { if ( err ) { reject ( err ) ; return } if ( response . errors ) { reject ( response . errors ) ; return } resolve ( response ) } ) } ) } ) this . log . trace ( 'Search reindex documents deleted' , model ) } catch ( e ) { this . log . trace ( "Search reindex destroy error" , model , e . message ) } for ( ; ; ) { let objs = await M . find ( ) . skip ( count ) . limit ( concurrent ) if ( objs . length === 0 ) break count += objs . length let docs = await Promise . all ( objs . map ( obj => this . _documentToIndex ( model , obj ) ) ) let body = _ . flatten ( docs . map ( ( d ) => [ { index : { _type , _id : d . id } } , d ] ) ) . map ( JSON . stringify ) . join ( "\n" ) try { await this . _retryLimit ( 'Reindex' , ( ) => { return new Promise ( ( resolve , reject ) => { SD . native ( ( err , client ) => { if ( err ) { reject ( err ) ; return } client . bulk ( { index , body } , ( err , response ) => { if ( err ) { reject ( err ) ; return } if ( response . errors ) { reject ( response . errors ) ; return } resolve ( response ) } ) } ) } ) } ) this . log . trace ( 'Bulk reindex progress' , count , 'documents indexed' ) } catch ( e ) { errors += 1 this . log . trace ( "Bulk reindex error" , e . message ) } if ( objs . length < concurrent ) break await timeout ( interval ) } this . log . trace ( 'Bulk reindex complete' , count , 'documents indexed' , "with" , errors , "errors" ) }
Reindex all of a model's documents.
Reindex all of a model ' s documents .
// the actual configured ES index and collection for native call // delete-by-query apparently does its own retries // bulk format is newline-separated action/data JSON strings // Although testing with ES 6.5 and _type deprecated since 5.x, validation error if `_type` is not provided //   the waterline-elasticsearch connection does set type too on non-native calls