<commit_before>var trigger    = require('level-trigger')
var liveStream = require('level-live-stream')
var viewStream = require('level-view-stream')
var delayJob   = require('./delay-job')

var Bucket  = require('range-bucket')
var map     = require('map-stream')
var through = require('through')

module.exports = function (db) {

  if(db.reduce) return

  trigger(db)
  liveStream(db)

  var views = {}
  db.reduce = {views: views}

  db.reduce.view = viewStream(db, db.reduce)

  db.reduce.add = function (view) {
    views[view.name] = view
    view.bucket = Bucket('mapr', view.name)
    view.depth = (view.depth && view.depth > 0) ? view.depth : 0
    var range = view.bucket.range()

    function doReduce (key, cb) {
      var collection = view.initial, values = []

      key = JSON.parse(key)
      key.push(true)

      db.readStream(view.bucket.range(key))
        .on('data', function (data) {
          collection = view.reduce(collection, data.value, data.key)
        })
        .on('end', function () {
          //save the collection
          //get the parent group
          var _key = key.slice(); _key.pop()
          db.put(view.bucket(_key), collection, cb)

          db.emit('reduce', view.name, _key, collection)
          db.emit('reduce:'+view.name, _key, collection)
        })
    }

    db.trigger.add({
      start: range.start,
      end  : range.end,
      map  : function (data) {
        var key = view.bucket.parse(data.key).key
        if(key.length <= view.depth) return

        key.pop()
        return JSON.stringify(key)
      }, 
      job  : doReduce
    })
  }
}<commit_msg>remove delayJob<commit_after>var trigger    = require('level-trigger')
var liveStream = require('level-live-stream')
var viewStream = require('level-view-stream')

var Bucket  = require('range-bucket')
var map     = require('map-stream')
var through = require('through')

module.exports = function (db) {

  if(db.reduce) return

  trigger(db)
  liveStream(db)

  var views = {}
  db.reduce = {views: views}

  db.reduce.view = viewStream(db, db.reduce)

  db.reduce.add = function (view) {
    views[view.name] = view
    view.bucket = Bucket('mapr', view.name)
    view.depth = (view.depth && view.depth > 0) ? view.depth : 0
    var range = view.bucket.range()

    function doReduce (key, cb) {
      var collection = view.initial, values = []

      key = JSON.parse(key)
      key.push(true)

      db.readStream(view.bucket.range(key))
        .on('data', function (data) {
          collection = view.reduce(collection, data.value, data.key)
        })
        .on('end', function () {
          //save the collection
          //get the parent group
          var _key = key.slice(); _key.pop()
          db.put(view.bucket(_key), collection, cb)

          db.emit('reduce', view.name, _key, collection)
          db.emit('reduce:'+view.name, _key, collection)
        })
    }

    db.trigger.add({
      start: range.start,
      end  : range.end,
      map  : function (data) {
        var key = view.bucket.parse(data.key).key
        if(key.length <= view.depth) return
        key.pop()
        return JSON.stringify(key)
      }, 
      job  : doReduce
    })
  }
}
{'fraction_non_alphanumeric': 0.14861067748985327, 'fraction_numerical': 0.001248829222603809, 'mean_word_length': 2.3375, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '6373931', 'n_tokens_mistral': 1040, 'n_tokens_neox': 1011, 'n_words': 226}