539021780

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/Rocketseat/adonis-bull/commit/<a class=hub.com/Rocketseat/adonis-bull/commit/<a class="double-link" href="https://git"><a class=hub.com/Rocketseat/adonis-bull/commit/<a class="double-link" href="https://git"><a class=hub.com/Rocketseat/adonis-bull/commit/<a class="double-link" href="https://git"><a class=hub.com/Rocketseat/adonis-bull/commit/902767be5ab59595b12cba91b328454770bf4a51">902767be5">&lt;a href=&quot;https://github.com/Rocketseat/adonis-bull/commit/</a><a class="double-link" href="https://github.com/Rocketseat/adonis-bull/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/Rocketseat/adonis-bull/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/Rocketseat/adonis-bull/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/Rocketseat/adonis-bull/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/Rocketseat/adonis-bull/commit/902767be5ab59595b12cba91b328454770bf4a51">902767be5</a><a href="https://github.com/Rocketseat/adonis-bull/commit/902767be5ab59595b12cba91b328454770bf4a51">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/Rocketseat/adonis-bull/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/Rocketseat/adonis-bull/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/Rocketseat/adonis-bull/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/Rocketseat/adonis-bull/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/Rocketseat/adonis-bull/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/Rocketseat/adonis-bull/commit/902767be5ab59595b12cba91b328454770bf4a51&quot;&gt;902767be5&lt;/a&gt;&lt;a href=&quot;https://github.com/Rocketseat/adonis-bull/commit/902767be5ab59595b12cba91b328454770bf4a51&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/Rocketseat/adonis-bull/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp... (continued)

Pull Request Pull Request #67: chore: Upgrade bull-board to 1.2.0

Run Details

15 of 21 branches covered (71.43%)

Branch coverage included in aggregate %.

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

69 of 97 relevant lines covered (71.13%)

2.82 hits per line

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

67.65

/src/Queue.js

'use strict'

const Bull = require('bull')
const BullBoard = require('bull-board')
const humanInterval = require('human-interval')

const differenceInMilliseconds = require('date-fns/differenceInMilliseconds')
const parseISO = require('date-fns/parseISO')
const fs = require('fs')

class Queue {
  constructor(Logger, Config, jobs, app, resolver) {
    this.Logger = Logger
    this.jobs = jobs

    this.app = app
    this.resolver = resolver

    this._queues = null

    const { connection, ...connections } = Config.get('bull')

    this.config = connections[connection]
    this.connections = connections
  }

  _getJobListeners(Job) {
    const jobListeners = Object.getOwnPropertyNames(Job.prototype)
      .filter((method) => method.startsWith('on'))
      .map((method) => {
        const eventName = method
          .replace(/^on(\w)/, (match, group) => group.toLowerCase())
          .replace(/([A-Z]+)/, (match, group) => ` ${group.toLowerCase()}`)

        return { eventName, method }
      })
    return jobListeners
  }

  get queues() {
    if (!this._queues) {
      this._queues = this.jobs.reduce((queues, path) => {
        const Job = this.app.use(path)

        let config = this.config
        if (Job.connection) {
          config = this.connections[Job.connection]
        }

        queues[Job.key] = {
          bull: new Bull(Job.key, config),
          Job,
          name: Job.key,
          handle: Job.handle,
          concurrency: Job.concurrency || 1,
          options: Job.options,
        }

        return queues
      }, {})
    }

    return this._queues
  }

  get(name) {
    return this.queues[name]
  }

  add(name, data, options) {
    const queue = this.get(name)

    const job = queue.bull.add(data, { ...queue.options, ...options })

    return job
  }

  schedule(name, data, date, options) {
    let delay

    if (typeof date === 'number' || date instanceof Number) {
      delay = date
    } else {
      if (typeof date === 'string' || date instanceof String) {
        const byHuman = humanInterval(date)
        if (!isNaN(byHuman)) {
          delay = byHuman
        } else {
          delay = differenceInMilliseconds(parseISO(date), new Date())
        }
      } else {
        delay = differenceInMilliseconds(date, new Date())
      }
    }

    if (delay > 0) {
      return this.add(name, data, { ...options, delay })
    } else {
      throw new Error('Invalid schedule time')
    }
  }

  ui(port = 9999) {
    BullBoard.setQueues(
      Object.values(this.queues).map(
        (queue) => new BullBoard.BullAdapter(queue.bull)
      )
    )

    const server = BullBoard.router.listen(port, () => {
      this.Logger.info(`bull board on http://localhost:${port}`)
    })

    const shutdown = () => {
      server.close(() => {
        this.Logger.info('Stopping bull board server')
        process.exit(0)
      })
    }

    process.on('SIGTERM', shutdown)
    process.on('SIGINT', shutdown)
  }

  async remove(name, jobId) {
    const job = await this.queues[name].bull.getJob(jobId)

    job.remove()
  }

  /* eslint handle-callback-err: "error" */
  handleException(error, job) {
    try {
      const exceptionHandlerFile = this.resolver
        .forDir('exceptions')
        .getPath('QueueHandler.js')
      fs.accessSync(exceptionHandlerFile, fs.constants.R_OK)

      const namespace = this.resolver
        .forDir('exceptions')
        .translate('QueueHandler')
      const handler = this.app.make(this.app.use(namespace))
      handler.report(error, job)
    } catch (err) {
      this.Logger.error(`name=${job.queue.name} id=${job.id}`)
    }
  }

  process() {
    this.Logger.info('Queue processing started')
    Object.values(this.queues).forEach((queue) => {
      const Job = new queue.Job()

      const jobListeners = this._getJobListeners(queue.Job)

      jobListeners.forEach(function (item) {
        queue.bull.on(item.eventName, Job[item.method].bind(Job))
      })

      queue.bull.process(queue.concurrency, (job, done) => {
        Job.handle(job)
          .then((result) => {
            done(null, result)
          })
          .catch((error) => {
            this.handleException(error, job)
            done(error)
          })
      })
    })

    const shutdown = () => {
      const promises = Object.values(this.queues).map((queue) => {
        return queue.bull.close()
      })

      return Promise.all(promises).then(process.exit(0))
    }

    process.on('SIGTERM', shutdown)
    process.on('SIGINT', shutdown)

    return this
  }
}

module.exports = Queue

1	'use strict'
2
3	const Bull = require('bull')	2✔
4	const BullBoard = require('bull-board')	2✔
5	const humanInterval = require('human-interval')	2✔
6
7	const differenceInMilliseconds = require('date-fns/differenceInMilliseconds')	2✔
8	const parseISO = require('date-fns/parseISO')	2✔
9	const fs = require('fs')	2✔
10
11	class Queue {
12	constructor(Logger, Config, jobs, app, resolver) {
13	this.Logger = Logger	10✔
14	this.jobs = jobs	10✔
15
16	this.app = app	10✔
17	this.resolver = resolver	10✔
18
19	this._queues = null	10✔
20
21	const { connection, ...connections } = Config.get('bull')	10✔
22
23	this.config = connections[connection]	10✔
24	this.connections = connections	10✔
25	}
26
27	_getJobListeners(Job) {
28	const jobListeners = Object.getOwnPropertyNames(Job.prototype)	2✔
29	.filter((method) => method.startsWith('on'))	6✔
30	.map((method) => {
31	const eventName = method	2✔
32	.replace(/^on(\w)/, (match, group) => group.toLowerCase())	2✔
33	.replace(/([A-Z]+)/, (match, group) => ` ${group.toLowerCase()}`)	×
34
35	return { eventName, method }	2✔
36	})
37	return jobListeners	2✔
38	}
39
40	get queues() {
41	if (!this._queues) {	10✔
42	this._queues = this.jobs.reduce((queues, path) => {	6✔
43	const Job = this.app.use(path)	6✔
44
45	let config = this.config	6✔
46	if (Job.connection) {	6!
47	config = this.connections[Job.connection]	×
48	}
49
50	queues[Job.key] = {	6✔
51	bull: new Bull(Job.key, config),
52	Job,
53	name: Job.key,
54	handle: Job.handle,
55	concurrency: Job.concurrency \|\| 1,	10✔
56	options: Job.options,
57	}
58
59	return queues	6✔
60	}, {})
61	}
62
63	return this._queues	10✔
64	}
65
66	get(name) {
67	return this.queues[name]	8✔
68	}
69
70	add(name, data, options) {
71	const queue = this.get(name)	6✔
72
73	const job = queue.bull.add(data, { ...queue.options, ...options })	6✔
74
75	return job	6✔
76	}
77
78	schedule(name, data, date, options) {
79	let delay
80
81	if (typeof date === 'number' \|\| date instanceof Number) {	4!
82	delay = date	×
83	} else {
84	if (typeof date === 'string' \|\| date instanceof String) {	4!
85	const byHuman = humanInterval(date)	4✔
86	if (!isNaN(byHuman)) {	4✔
87	delay = byHuman	2✔
88	} else {
89	delay = differenceInMilliseconds(parseISO(date), new Date())	2✔
90	}
91	} else {
92	delay = differenceInMilliseconds(date, new Date())	×
93	}
94	}
95
96	if (delay > 0) {	4✔
97	return this.add(name, data, { ...options, delay })	2✔
98	} else {
99	throw new Error('Invalid schedule time')	2✔
100	}
101	}
102
103	ui(port = 9999) {	×
104	BullBoard.setQueues(	×
105	Object.values(this.queues).map(
106	(queue) => new BullBoard.BullAdapter(queue.bull)	×
107	)
108	)
109
110	const server = BullBoard.router.listen(port, () => {	×
111	this.Logger.info(`bull board on http://localhost:${port}`)	×
112	})
113
114	const shutdown = () => {	×
115	server.close(() => {	×
116	this.Logger.info('Stopping bull board server')	×
117	process.exit(0)	×
118	})
119	}
120
121	process.on('SIGTERM', shutdown)	×
122	process.on('SIGINT', shutdown)	×
123	}
124
125	async remove(name, jobId) {
126	const job = await this.queues[name].bull.getJob(jobId)	×
127
128	job.remove()	×
129	}
130
131	/* eslint handle-callback-err: "error" */
132	handleException(error, job) {
133	try {	×
134	const exceptionHandlerFile = this.resolver	×
135	.forDir('exceptions')
136	.getPath('QueueHandler.js')
137	fs.accessSync(exceptionHandlerFile, fs.constants.R_OK)	×
138
139	const namespace = this.resolver	×
140	.forDir('exceptions')
141	.translate('QueueHandler')
142	const handler = this.app.make(this.app.use(namespace))	×
143	handler.report(error, job)	×
144	} catch (err) {
145	this.Logger.error(`name=${job.queue.name} id=${job.id}`)	×
146	}
147	}
148
149	process() {
150	this.Logger.info('Queue processing started')	2✔
151	Object.values(this.queues).forEach((queue) => {	2✔
152	const Job = new queue.Job()	2✔
153
154	const jobListeners = this._getJobListeners(queue.Job)	2✔
155
156	jobListeners.forEach(function (item) {	2✔
157	queue.bull.on(item.eventName, Job[item.method].bind(Job))	2✔
158	})
159
160	queue.bull.process(queue.concurrency, (job, done) => {	2✔
161	Job.handle(job)	2✔
162	.then((result) => {
163	done(null, result)	2✔
164	})
165	.catch((error) => {
166	this.handleException(error, job)	×
167	done(error)	×
168	})
169	})
170	})
171
172	const shutdown = () => {	2✔
173	const promises = Object.values(this.queues).map((queue) => {	×
174	return queue.bull.close()	×
175	})
176
177	return Promise.all(promises).then(process.exit(0))	×
178	}
179
180	process.on('SIGTERM', shutdown)	2✔
181	process.on('SIGINT', shutdown)	2✔
182
183	return this	2✔
184	}
185	}
186
187	module.exports = Queue	2✔

Rocketseat / adonis-bull / 539021780

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