TrinetraLabs/Placebo_AI

1var inherits = require('inherits')2var EventEmitter = require('events').EventEmitter3 4module.exports = Queue5module.exports.default = Queue6 7function Queue (options) {8  if (!(this instanceof Queue)) {9    return new Queue(options)10  }11 12  EventEmitter.call(this)13  options = options || {}14  this.concurrency = options.concurrency || Infinity15  this.timeout = options.timeout || 016  this.autostart = options.autostart || false17  this.results = options.results || null18  this.pending = 019  this.session = 020  this.running = false21  this.jobs = []22  this.timers = {}23}24inherits(Queue, EventEmitter)25 26var arrayMethods = [27  'pop',28  'shift',29  'indexOf',30  'lastIndexOf'31]32 33arrayMethods.forEach(function (method) {34  Queue.prototype[method] = function () {35    return Array.prototype[method].apply(this.jobs, arguments)36  }37})38 39Queue.prototype.slice = function (begin, end) {40  this.jobs = this.jobs.slice(begin, end)41  return this42}43 44Queue.prototype.reverse = function () {45  this.jobs.reverse()46  return this47}48 49var arrayAddMethods = [50  'push',51  'unshift',52  'splice'53]54 55arrayAddMethods.forEach(function (method) {56  Queue.prototype[method] = function () {57    var methodResult = Array.prototype[method].apply(this.jobs, arguments)58    if (this.autostart) {59      this.start()60    }61    return methodResult62  }63})64 65Object.defineProperty(Queue.prototype, 'length', {66  get: function () {67    return this.pending + this.jobs.length68  }69})70 71Queue.prototype.start = function (cb) {72  if (cb) {73    callOnErrorOrEnd.call(this, cb)74  }75 76  this.running = true77 78  if (this.pending >= this.concurrency) {79    return80  }81 82  if (this.jobs.length === 0) {83    if (this.pending === 0) {84      done.call(this)85    }86    return87  }88 89  var self = this90  var job = this.jobs.shift()91  var once = true92  var session = this.session93  var timeoutId = null94  var didTimeout = false95  var resultIndex = null96  var timeout = job.hasOwnProperty('timeout') ? job.timeout : this.timeout97 98  function next (err, result) {99    if (once && self.session === session) {100      once = false101      self.pending--102      if (timeoutId !== null) {103        delete self.timers[timeoutId]104        clearTimeout(timeoutId)105      }106 107      if (err) {108        self.emit('error', err, job)109      } else if (didTimeout === false) {110        if (resultIndex !== null) {111          self.results[resultIndex] = Array.prototype.slice.call(arguments, 1)112        }113        self.emit('success', result, job)114      }115 116      if (self.session === session) {117        if (self.pending === 0 && self.jobs.length === 0) {118          done.call(self)119        } else if (self.running) {120          self.start()121        }122      }123    }124  }125 126  if (timeout) {127    timeoutId = setTimeout(function () {128      didTimeout = true129      if (self.listeners('timeout').length > 0) {130        self.emit('timeout', next, job)131      } else {132        next()133      }134    }, timeout)135    this.timers[timeoutId] = timeoutId136  }137 138  if (this.results) {139    resultIndex = this.results.length140    this.results[resultIndex] = null141  }142 143  this.pending++144  self.emit('start', job)145  var promise = job(next)146  if (promise && promise.then && typeof promise.then === 'function') {147    promise.then(function (result) {148      return next(null, result)149    }).catch(function (err) {150      return next(err || true)151    })152  }153 154  if (this.running && this.jobs.length > 0) {155    this.start()156  }157}158 159Queue.prototype.stop = function () {160  this.running = false161}162 163Queue.prototype.end = function (err) {164  clearTimers.call(this)165  this.jobs.length = 0166  this.pending = 0167  done.call(this, err)168}169 170function clearTimers () {171  for (var key in this.timers) {172    var timeoutId = this.timers[key]173    delete this.timers[key]174    clearTimeout(timeoutId)175  }176}177 178function callOnErrorOrEnd (cb) {179  var self = this180  this.on('error', onerror)181  this.on('end', onend)182 183  function onerror (err) { self.end(err) }184  function onend (err) {185    self.removeListener('error', onerror)186    self.removeListener('end', onend)187    cb(err, this.results)188  }189}190 191function done (err) {192  this.session++193  this.running = false194  this.emit('end', err)195}196