<commit_before>var Obv = require('obv')
var Drain = require('pull-stream/sinks/drain')
var AtomicFile = require('atomic-file')
var path = require('path')
var deepEqual = require('deep-equal')

module.exports = function (reduce) {
  return function (log, name) { //name is where this view is mounted
    var acc, since = Obv()
    var value = Obv(), _value

    // if the log is persisted,
    // then also save the reduce state.
    // currently saving it every minute, if it's changed.
    // I don't think this is the best way
    // but it's the easiest...
    if(log.dir) {
      var state = AtomicFile(path.join(log.dir, name+'.json'))
      state.get(function (err, data) {
        if(err) since.set(-1)
        else {
          value.set(_value = data.value)
          since.set(data.seq)
        }
      })
      ;(function next() {
        setTimeout(function () {
          if(since.value > 0 && !deepEqual(_value, value.value))
            state.set({seq: since.value, value: _value = value.value}, function () {})
          next()
        }, 1000*60).unref()
      })()
    }
    else
      since.set(-1)

    return {
      since: since,
      value: value,
      methods: {get: 'async'},
      get: function (path, cb) {
        cb(null, value.value)
      },
      createSink: function (cb) {
        return Drain(function (data) {
          value.set(reduce(value.value, data.value, data.seq))
          since.set(data.seq)
        }, cb)
      },
      destroy: function (cb) {
        value.set(null); since.set(-1);
        if(state) state.set(null, cb)
        else cb()
      }
    }
  }
}

<commit_msg>flush any writes on close<commit_after>var Obv = require('obv')
var Drain = require('pull-stream/sinks/drain')
var AtomicFile = require('atomic-file')
var path = require('path')
var deepEqual = require('deep-equal')

function isEmpty (o) {
  for(var k in o) return false
  return true
}

module.exports = function (reduce) {
  return function (log, name) { //name is where this view is mounted
    var acc, since = Obv()
    var value = Obv(), _value, int

    // if the log is persisted,
    // then also save the reduce state.
    // currently saving it every minute, if it's changed.
    // I don't think this is the best way
    // but it's the easiest...
    if(log.filename) {
      var dir = path.dirname(log.filename)
      var state = AtomicFile(path.join(dir, name+'.json'))
      state.get(function (err, data) {
        if(err || isEmpty(data)) since.set(-1)
        else {
          value.set(_value = data.value)
          since.set(data.seq)
        }
      })
      ;(function next() {
        int = setTimeout(function () {
          if(since.value > 0 && !deepEqual(_value, value.value))
            state.set({seq: since.value, value: _value = value.value}, function () {})
          next()
        }, 1000*60)
        int.unref()
      })()
    }
    else
      since.set(-1)

    return {
      since: since,
      value: value,
      methods: {get: 'async'},
      get: function (path, cb) {
        cb(null, value.value)
      },
      createSink: function (cb) {
        return Drain(function (data) {
          value.set(reduce(value.value, data.value, data.seq))
          since.set(data.seq)
        }, cb)
      },
      destroy: function (cb) {
        value.set(null); since.set(-1);
        if(state) state.set(null, cb)
        else cb()
      },
      close: function (cb) {
        clearTimeout(int)
        if(!since.value) return cb()
        //force a write.
        state.set({seq: since.value, value: _value = value.value}, cb)
      }
    }
  }
}






{'fraction_non_alphanumeric': 0.14629424778761063, 'fraction_numerical': 0.0055309734513274336, 'mean_word_length': 2.0191986644407347, 'pattern_counts': {'":': 0, '<': 3, '<?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': '3149645', 'n_tokens_mistral': 1165, 'n_tokens_neox': 1134, 'n_words': 306}