Ushio-cn-old/yimian/iot/mqtt/smartfarm/node_modules/pull-window/index.js

var looper = require('looper')

var window = module.exports = function (init, start) {
return function (read) {
  start = start || function (start, data) {
    return {start: start, data: data}
  }
  var windows = [], output = [], ended = null
  var data, end
  var j = 0

  return function (abort, cb) {
    if(output.length)
      return cb(null, output.shift())
    if(ended)
      return cb(ended)
    var i = 0
    var k = j ++
    read(abort, looper(function (end, data) {
      var next = this
      var reduce, update, once = false
      if(end)
        ended = end

      function _update (end, _data) {
        if(once) return
        once = true
        delete windows[windows.indexOf(update)]
        output.push(start(data, _data))
      }

      if(!ended)
        update = init(data, _update)

      if(update)
        windows.push(update)
      else
        //don't allow data unless a window started here!
        once = true

      windows.forEach(function (update, i) {
        update(end, data)
      })

      if(output.length)
        return cb(null, output.shift())
      else if(ended)
        return cb(ended)
      else
        read(null, next)

  }))
  }
}}

window.recent = function (size, time) {
  var current = null
  return window(function (data, cb) {
    if(current) return
    current = []
    var timer
      
    function done () {
      var _current = current
      current = null
      clearTimeout(timer)
      cb(null, _current)
    }

    if(time)
      timer = setTimeout(done, time)

    return function (end, data) {
      if(end) return done()
      current.push(data)
      if(size != null && current.length >= size)
        done()
    }
  }, function (_, data) {
    return data
  })
}

window.sliding = function (reduce, width) {
  width = width || 10
  var k = 0
  return window(function (data, cb) {
    var acc
    var i = 0
    var l = k++
    return function (end, data) {
      if(end) return
      acc = reduce(acc, data)
      if(width <= ++ i)
        cb(null, acc)
    }
  })
}