queue

Asynchronous function queue with adjustable concurrency

Queue
jessetane/queuequeue

README

  1. ```
  2.    ____  __  _____  __  _____
  3.   / __ `/ / / / _ \/ / / / _ \
  4. / /_/ / /_/ /  __/ /_/ /  __/
  5. \__, /\__,_/\___/\__,_/\___/
  6.    /_/

  8. ```

Asynchronous function queue with adjustable concurrency.

This module exports a class Queuethat implements most of the ArrayAPI. Pass async functions (ones that accept a callback or return a promise) to an instance's additive array methods. Processing begins when you call q.start().

Example


Do npm run exampleor npm run devand open the example directory (and your console) to run the following program:

  1. ``` js
  2. import Queue from 'queue'

  4. const q = new Queue({ results: [] })

  6. // add jobs using the familiar Array API
  7. q.push(cb => {
  8.   const result = 'two'
  9.   cb(null, result)
  10. })

  12. q.push(
  13.   cb => {
  14.     const result = 'four'
  15.     cb(null, result)
  16.   },
  17.   cb => {
  18.     const result = 'five'
  19.     cb(null, result)
  20.   }
  21. )

  23. // jobs can accept a callback or return a promise
  24. q.push(() => {
  25.   return new Promise((resolve, reject) => {
  26.     const result = 'one'
  27.     resolve(result)
  28.   })
  29. })

  31. q.unshift(cb => {
  32.   const result = 'one'
  33.   cb(null, result)
  34. })

  36. q.splice(2, 0, cb => {
  37.   const result = 'three'
  38.   cb(null, result)
  39. })

  41. // use the timeout feature to deal with jobs that
  42. // take too long or forget to execute a callback
  43. q.timeout = 100

  45. q.addEventListener('timeout', e => {
  46.   console.log('job timed out:', e.detail.job.toString().replace(/\n/g, ''))
  47.   e.detail.next()
  48. })

  50. q.push(cb => {
  51.   setTimeout(() => {
  52.     console.log('slow job finished')
  53.     cb()
  54.   }, 200)
  55. })

  57. q.push(cb => {
  58.   console.log('forgot to execute callback')
  59. })

  61. // jobs can also override the queue's timeout
  62. // on a per-job basis
  63. function extraSlowJob (cb) {
  64.   setTimeout(() => {
  65.     console.log('extra slow job finished')
  66.     cb()
  67.   }, 400)
  68. }

  70. extraSlowJob.timeout = 500
  71. q.push(extraSlowJob)

  73. // jobs can also opt-out of the timeout altogether
  74. function superSlowJob (cb) {
  75.   setTimeout(() => {
  76.     console.log('super slow job finished')
  77.     cb()
  78.   }, 1000)
  79. }

  81. superSlowJob.timeout = null
  82. q.push(superSlowJob)

  84. // get notified when jobs complete
  85. q.addEventListener('success', e => {
  86.   console.log('job finished processing:', e.detail.toString().replace(/\n/g, ''))
  87.   console.log('The result is:', e.detail.result)
  88. })

  90. // begin processing, get notified on end / failure
  91. q.start(err => {
  92.   if (err) throw err
  93.   console.log('all done:', q.results)
  94. })
  95. ```

Install


  1. ``` null
  2. npm install queue

  4. yarn add queue

  6. ```

Test


  1. ``` null
  2. npm test

  4. npm run dev // for testing in a browser, open test directory (and your console)

  6. ```

API


const q = new Queue([opts])


Constructor. optsmay contain initial values for:

q.concurrency
q.timeout
q.autostart
q.results

Instance methods


q.start([cb])


Explicitly starts processing jobs and provides feedback to the caller when the queue empties or an error occurs. If cb is not passed a promise will be returned.

q.stop()


Stops the queue. can be resumed with q.start().

q.end([err])


Stop and empty the queue immediately.

Instance methods mixed in from Array


Mozilla has docs on how these methods work here . Note that slicedoes not copy the queue.

q.push(element1, ..., elementN)


q.unshift(element1, ..., elementN)


q.splice(index , howMany[, element1[, ...[, elementN]]])


q.pop()


q.shift()


q.slice(begin[, end])


q.reverse()


q.indexOf(searchElement[, fromIndex])


q.lastIndexOf(searchElement[, fromIndex])


Properties


q.concurrency


Max number of jobs the queue should process concurrently, defaults to Infinity.

q.timeout


Milliseconds to wait for a job to execute its callback. This can be overridden by specifying a timeoutproperty on a per-job basis.

q.autostart


Ensures the queue is always running if jobs are available. Useful in situations where you are using a queue only for concurrency control.

q.results


An array to set job callback arguments on.

q.length


Jobs pending + jobs to process (readonly).

Events


q.dispatchEvent(new QueueEvent('start', { job }))


Immediately before a job begins to execute.

q.dispatchEvent(new QueueEvent('success', { result: [...result], job }))


After a job executes its callback.

q.dispatchEvent(new QueueEvent('error', { err, job }))


After a job passes an error to its callback.

q.dispatchEvent(new QueueEvent('timeout', { next, job }))


After q.timeoutmilliseconds have elapsed and a job has not executed its callback.

q.dispatchEvent(new QueueEvent('end', { err }))


After all jobs have been processed

Releases


The latest stable release is published to npm . Abbreviated changelog below:

7.0
Modernized codebase, added new maintainer (@MaksimLavrenyuk)

6.0
Add startevent before job begins (@joelgriffith)
Add timeoutproperty on a job to override the queue's timeout (@joelgriffith)

5.0
Updated TypeScript bindings (@Codex-)

4.4
Add results feature

4.3
Add promise support (@kwolfy)

4.2
Unref timers on end

4.1
Add autostart feature

4.0
Change license to MIT

3.1.x
Add .npmignore

3.0.x
Change the default concurrency to Infinity
Allow q.start()to accept an optional callback executed on q.emit('end')

2.x
Major api changes / not backwards compatible with 1.x

1.x
Early prototype

License


MIT