strong-tie/inbound-calls

1'use strict'2 3/* eslint-disable no-var */4 5var reusify = require('reusify')6 7function fastqueue (context, worker, _concurrency) {8  if (typeof context === 'function') {9    _concurrency = worker10    worker = context11    context = null12  }13 14  if (!(_concurrency >= 1)) {15    throw new Error('fastqueue concurrency must be equal to or greater than 1')16  }17 18  var cache = reusify(Task)19  var queueHead = null20  var queueTail = null21  var _running = 022  var errorHandler = null23 24  var self = {25    push: push,26    drain: noop,27    saturated: noop,28    pause: pause,29    paused: false,30 31    get concurrency () {32      return _concurrency33    },34    set concurrency (value) {35      if (!(value >= 1)) {36        throw new Error('fastqueue concurrency must be equal to or greater than 1')37      }38      _concurrency = value39 40      if (self.paused) return41      for (; queueHead && _running < _concurrency;) {42        _running++43        release()44      }45    },46 47    running: running,48    resume: resume,49    idle: idle,50    length: length,51    getQueue: getQueue,52    unshift: unshift,53    empty: noop,54    kill: kill,55    killAndDrain: killAndDrain,56    error: error57  }58 59  return self60 61  function running () {62    return _running63  }64 65  function pause () {66    self.paused = true67  }68 69  function length () {70    var current = queueHead71    var counter = 072 73    while (current) {74      current = current.next75      counter++76    }77 78    return counter79  }80 81  function getQueue () {82    var current = queueHead83    var tasks = []84 85    while (current) {86      tasks.push(current.value)87      current = current.next88    }89 90    return tasks91  }92 93  function resume () {94    if (!self.paused) return95    self.paused = false96    if (queueHead === null) {97      _running++98      release()99      return100    }101    for (; queueHead && _running < _concurrency;) {102      _running++103      release()104    }105  }106 107  function idle () {108    return _running === 0 && self.length() === 0109  }110 111  function push (value, done) {112    var current = cache.get()113 114    current.context = context115    current.release = release116    current.value = value117    current.callback = done || noop118    current.errorHandler = errorHandler119 120    if (_running >= _concurrency || self.paused) {121      if (queueTail) {122        queueTail.next = current123        queueTail = current124      } else {125        queueHead = current126        queueTail = current127        self.saturated()128      }129    } else {130      _running++131      worker.call(context, current.value, current.worked)132    }133  }134 135  function unshift (value, done) {136    var current = cache.get()137 138    current.context = context139    current.release = release140    current.value = value141    current.callback = done || noop142    current.errorHandler = errorHandler143 144    if (_running >= _concurrency || self.paused) {145      if (queueHead) {146        current.next = queueHead147        queueHead = current148      } else {149        queueHead = current150        queueTail = current151        self.saturated()152      }153    } else {154      _running++155      worker.call(context, current.value, current.worked)156    }157  }158 159  function release (holder) {160    if (holder) {161      cache.release(holder)162    }163    var next = queueHead164    if (next && _running <= _concurrency) {165      if (!self.paused) {166        if (queueTail === queueHead) {167          queueTail = null168        }169        queueHead = next.next170        next.next = null171        worker.call(context, next.value, next.worked)172        if (queueTail === null) {173          self.empty()174        }175      } else {176        _running--177      }178    } else if (--_running === 0) {179      self.drain()180    }181  }182 183  function kill () {184    queueHead = null185    queueTail = null186    self.drain = noop187  }188 189  function killAndDrain () {190    queueHead = null191    queueTail = null192    self.drain()193    self.drain = noop194  }195 196  function error (handler) {197    errorHandler = handler198  }199}200 201function noop () {}202 203function Task () {204  this.value = null205  this.callback = noop206  this.next = null207  this.release = noop208  this.context = null209  this.errorHandler = null210 211  var self = this212 213  this.worked = function worked (err, result) {214    var callback = self.callback215    var errorHandler = self.errorHandler216    var val = self.value217    self.value = null218    self.callback = noop219    if (self.errorHandler) {220      errorHandler(err, val)221    }222    callback.call(self.context, err, result)223    self.release(self)224  }225}226 227function queueAsPromised (context, worker, _concurrency) {228  if (typeof context === 'function') {229    _concurrency = worker230    worker = context231    context = null232  }233 234  function asyncWrapper (arg, cb) {235    worker.call(this, arg)236      .then(function (res) {237        cb(null, res)238      }, cb)239  }240 241  var queue = fastqueue(context, asyncWrapper, _concurrency)242 243  var pushCb = queue.push244  var unshiftCb = queue.unshift245 246  queue.push = push247  queue.unshift = unshift248  queue.drained = drained249 250  return queue251 252  function push (value) {253    var p = new Promise(function (resolve, reject) {254      pushCb(value, function (err, result) {255        if (err) {256          reject(err)257          return258        }259        resolve(result)260      })261    })262 263    // Let's fork the promise chain to264    // make the error bubble up to the user but265    // not lead to a unhandledRejection266    p.catch(noop)267 268    return p269  }270 271  function unshift (value) {272    var p = new Promise(function (resolve, reject) {273      unshiftCb(value, function (err, result) {274        if (err) {275          reject(err)276          return277        }278        resolve(result)279      })280    })281 282    // Let's fork the promise chain to283    // make the error bubble up to the user but284    // not lead to a unhandledRejection285    p.catch(noop)286 287    return p288  }289 290  function drained () {291    var p = new Promise(function (resolve) {292      process.nextTick(function () {293        if (queue.idle()) {294          resolve()295        } else {296          var previousDrain = queue.drain297          queue.drain = function () {298            if (typeof previousDrain === 'function') previousDrain()299            resolve()300            queue.drain = previousDrain301          }302        }303      })304    })305 306    return p307  }308}309 310module.exports = fastqueue311module.exports.promise = queueAsPromised312