<commit_before>"use strict";

const through2 = require('through2')
    , BatchStream = require('batch-stream')
    , duplexify = require('duplexify')

module.exports = function levelBatchStream(db, opts = {}) {
  const { silent, size = 100, type, ...batchOpts } = opts

  const ws = new BatchStream({ size })

  const rs = through2.obj(function(ops, _, next) {
    doBatch(ops, function(err) {
      if (err) {
        dup.emit('batch-error', err, ops)
        if (!silent) return dup.destroy(err)
      }

      next()
    })
  })

  function doBatch(ops, next) {
    if (type) {
      ops = ops.map(function(op){
        op.type = op.type || type
        return op
      })
    }

    db.batch(ops, batchOpts, next)
  }

  ws.pipe(rs)

  // TODO: use pumpify
  var dup = duplexify.obj(ws, rs)

  dup.clearBatch = function() {
    return ws.batch.splice(0, ws.batch.length)
  }

  dup.getBatch = function() {
    return ws.batch.slice()
  }

  // TODO: include ops buffered in rs?
  dup.flushBatch = function(cb) {
    const batch = dup.clearBatch()
    doBatch(batch, cb)
  }

  return dup
}<commit_msg>batch stream with timeout<commit_after>"use strict";

const through2 = require('through2')

module.exports = function levelBatchStream(db, opts = {}) {
  const { silent, type, highWaterMark, ...batchOpts,  } = opts
  const batch = []

  let timeout;

  const stream = through2.obj({ highWaterMark }, function (op, _, next) {
    if (type && !op.type) op.type = type
    batch.push(op)

    if (timeout == null) {
      timeout = setTimeout(function(){
        timeout = null
        flushBatch()
      }, 500)
    }

    next()
  }, function flush(cb) {
    clearTimeout(timeout)
    timeout = null
    flushBatch(cb)
  })

  function flushBatch(cb) {
    if (batch.length) {
      const ops = batch.splice(0, batch.length)

      db.batch(ops, batchOpts, function(err) {
        if (err) {
          stream.emit('batch-error', err, ops)
          if (!silent) return stream.destroy(err)
        }

        cb && cb()
      })
    } else {
      cb && cb()
    }
  }

  return stream
}
{'fraction_non_alphanumeric': 0.14545454545454545, 'fraction_numerical': 0.0066985645933014355, 'mean_word_length': 2.3944805194805197, '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': '5961451', 'n_tokens_mistral': 716, 'n_tokens_neox': 688, 'n_words': 166}