e-e-e / a-promise-queue / 14

pending completion

Build # 14

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #1 from e-e-e/next

Change interface and add promise like behavior to PromiseQueue

Run Details

56 of 61 branches covered (91.8%)

Branch coverage included in aggregate %.

18 of 18 new or added lines in 1 file covered. (100.0%)

72 of 73 relevant lines covered (98.63%)

89.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.52

/index.js

class PromiseQueue {
  constructor (options, callback) {
    const cb = (typeof options === 'function') ? options : callback
    const opts = (options && typeof options === 'object') ? options : {}
    this.flushing = false
    this.Promise = opts.promise || Promise
    this.concurrency = (!opts.concurrency || typeof opts.concurrency !== 'number') ? 1 : opts.concurrency
    this.promises = []
    this.queue = []
    this.callback = cb
    if (!cb) {
      this._makePromise()
    }
  }

  _makePromise () {
    this._finalPromise = new Promise((resolve, reject) => {
      this.resolve = resolve
      this.reject = reject
    })
  }

  then (onResolve, onReject) {
    if (this.callback) throw new Error('Cannot use PromiseQueue as a Promise if callback has been is set')
    return this._finalPromise.then(onResolve, onReject)
  }

  catch (onReject) {
    if (this.callback) throw new Error('Cannot use PromiseQueue as a Promise if callback has been is set')
    return this._finalPromise.catch(onReject)
  }

  add (fn, opts) {
    if (typeof fn !== 'function') throw new Error('PromiseQueue.add() expects a function as an argument.')
    return new this.Promise((resolve, reject) => {
      const attempts = (opts && opts.attempts && opts.attempts > 0) ? opts.attempts : 1
      if (this.promises.length < this.concurrency) {
        const id = (this.promises.length) ? this.promises[this.promises.length - 1].id + 1 : 1
        this.promises.push({
          id,
          promise: this._wrap(fn, id, resolve, reject, attempts)
        })
      } else {
        // shift order based on priority
        const next = {
          fn,
          attempts,
          priority: (opts && opts.priority) ? opts.priority : 0,
          resolve,
          reject
        }
        if (!opts || !opts.priority) {
          this.queue.push(next)
        } else {
          let found = false
          for (let i = this.length - 1; i >= 0; i--) {
            if (this.queue[i].priority && this.queue[i].priority >= opts.priority) {
              this.queue.splice(i + 1, 0, next)
              found = true
              break
            }
          }
          if (!found) {
            this.queue.unshift(next)
          }
        }
      }
    })
  }

  flush () {
    const currentPromises = this.promises.map(p => p.promise)
    const concurrent = [...currentPromises, ...this.queue.map(queued => this._promised(queued.fn).then(queued.resolve, queued.reject))]
    this.flushing = true
    this.queue = []

    const flushed = () => {
      this.flushing = false
      if (this.length > 0) {
        // start processing new additions
        const nextFn = this.queue.shift()
        const id = (this.promises.length) ? this.promises[this.promises.length - 1].id + 1 : 1
        const promise = this._wrap(nextFn.fn, id, nextFn.resolve, nextFn.reject, nextFn.attempts)
        this.promises.push({ id, promise })
      }
    }

    return this.Promise.all(concurrent)
      .then(flushed, flushed)
  }

  get length () {
    return this.queue.length
  }

  _promised (fn) {
    try {
      return this.Promise.resolve(fn())
    } catch (e) {
      return this.Promise.reject(e)
    }
  }

  _next (id) {
    if (this.flushing) return
    if (this.length > 0) {
      const nextFn = this.queue.shift()
      return this._wrap(nextFn.fn, id, nextFn.resolve, nextFn.reject, nextFn.attempts)
    }
    const promiseId = this.promises.findIndex(p => {
      return p.id === id
    })
    this.promises.splice(promiseId, 1)
    if (this.promises.length === 0) this._done()
    return true
  }

  _wrap (fn, id, resolve, reject, attempts) {
    let retryCount = 0
    const retry = (err) => {
      if (retryCount >= attempts) {
        throw err || new Error('Unknown Error')
      }
      retryCount += 1
      return this._promised(fn).catch(retry)
    }
    return retry()
      .then((r) => { resolve(r) }, (e) => { reject(e) })
      .then(() => this._next(id))
  }

  _done (err) {
    if (typeof this.callback === 'function') this.callback(err)
    else if (this._finalPromise) {
      this.resolve()
      this._makePromise()
    }
  }
}

module.exports = PromiseQueue

1	class PromiseQueue {
2	constructor (options, callback) {
3	const cb = (typeof options === 'function') ? options : callback	68✔
4	const opts = (options && typeof options === 'object') ? options : {}	68✔
5	this.flushing = false	68✔
6	this.Promise = opts.promise \|\| Promise	68✔
7	this.concurrency = (!opts.concurrency \|\| typeof opts.concurrency !== 'number') ? 1 : opts.concurrency	68✔
8	this.promises = []	68✔
9	this.queue = []	68✔
10	this.callback = cb	68✔
11	if (!cb) {	68✔
12	this._makePromise()	36✔
13	}
14	}
15
16	_makePromise () {
17	this._finalPromise = new Promise((resolve, reject) => {	64✔
18	this.resolve = resolve	64✔
19	this.reject = reject	64✔
20	})
21	}
22
23	then (onResolve, onReject) {
24	if (this.callback) throw new Error('Cannot use PromiseQueue as a Promise if callback has been is set')	16✔
25	return this._finalPromise.then(onResolve, onReject)	12✔
26	}
27
28	catch (onReject) {
29	if (this.callback) throw new Error('Cannot use PromiseQueue as a Promise if callback has been is set')	4!
30	return this._finalPromise.catch(onReject)	×
31	}
32
33	add (fn, opts) {
34	if (typeof fn !== 'function') throw new Error('PromiseQueue.add() expects a function as an argument.')	240✔
35	return new this.Promise((resolve, reject) => {	236✔
36	const attempts = (opts && opts.attempts && opts.attempts > 0) ? opts.attempts : 1	236✔
37	if (this.promises.length < this.concurrency) {	236✔
38	const id = (this.promises.length) ? this.promises[this.promises.length - 1].id + 1 : 1	76✔
39	this.promises.push({	76✔
40	id,
41	promise: this._wrap(fn, id, resolve, reject, attempts)
42	})
43	} else {
44	// shift order based on priority
45	const next = {	160✔
46	fn,
47	attempts,
48	priority: (opts && opts.priority) ? opts.priority : 0,	344✔
49	resolve,
50	reject
51	}
52	if (!opts \|\| !opts.priority) {	160✔
53	this.queue.push(next)	148✔
54	} else {
55	let found = false	12✔
56	for (let i = this.length - 1; i >= 0; i--) {	12✔
57	if (this.queue[i].priority && this.queue[i].priority >= opts.priority) {	32✔
58	this.queue.splice(i + 1, 0, next)	4✔
59	found = true	4✔
60	break	4✔
61	}
62	}
63	if (!found) {	12✔
64	this.queue.unshift(next)	8✔
65	}
66	}
67	}
68	})
69	}
70
71	flush () {
72	const currentPromises = this.promises.map(p => p.promise)	4✔
73	const concurrent = [...currentPromises, ...this.queue.map(queued => this._promised(queued.fn).then(queued.resolve, queued.reject))]	12✔
74	this.flushing = true	4✔
75	this.queue = []	4✔
76
77	const flushed = () => {	4✔
78	this.flushing = false	4✔
79	if (this.length > 0) {	4!
80	// start processing new additions
81	const nextFn = this.queue.shift()	4✔
82	const id = (this.promises.length) ? this.promises[this.promises.length - 1].id + 1 : 1	4!
83	const promise = this._wrap(nextFn.fn, id, nextFn.resolve, nextFn.reject, nextFn.attempts)	4✔
84	this.promises.push({ id, promise })	4✔
85	}
86	}
87
88	return this.Promise.all(concurrent)	4✔
89	.then(flushed, flushed)
90	}
91
92	get length () {
93	return this.queue.length	256✔
94	}
95
96	_promised (fn) {
97	try {	272✔
98	return this.Promise.resolve(fn())	272✔
99	} catch (e) {
100	return this.Promise.reject(e)	4✔
101	}
102	}
103
104	_next (id) {
105	if (this.flushing) return	224✔
106	if (this.length > 0) {	220✔
107	const nextFn = this.queue.shift()	144✔
108	return this._wrap(nextFn.fn, id, nextFn.resolve, nextFn.reject, nextFn.attempts)	144✔
109	}
110	const promiseId = this.promises.findIndex(p => {	76✔
111	return p.id === id	80✔
112	})
113	this.promises.splice(promiseId, 1)	76✔
114	if (this.promises.length === 0) this._done()	76✔
115	return true	76✔
116	}
117
118	_wrap (fn, id, resolve, reject, attempts) {
119	let retryCount = 0	224✔
120	const retry = (err) => {	224✔
121	if (retryCount >= attempts) {	276✔
122	throw err \|\| new Error('Unknown Error')	16!
123	}
124	retryCount += 1	260✔
125	return this._promised(fn).catch(retry)	260✔
126	}
127	return retry()	224✔
128	.then((r) => { resolve(r) }, (e) => { reject(e) })	208✔
129	.then(() => this._next(id))	224✔
130	}
131
132	_done (err) {
133	if (typeof this.callback === 'function') this.callback(err)	56✔
134	else if (this._finalPromise) {	28!
135	this.resolve()	28✔
136	this._makePromise()	28✔
137	}
138	}
139	}
140
141	module.exports = PromiseQueue	4✔

e-e-e / a-promise-queue / 14

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous