<commit_before>import './Job'
import './ConfigBuilder'
import './QueryBuilder'

import Kue from 'kue'

export class Queue {
	kue = null
	app = null
	jobs = { }

	constructor(app, config) {
		this.app = app
		this.kue = new Kue(ConfigBuilder(config, app))
	}

	register(jobClass) {
		if(jobClass.jobName.isNil) {
			throw new Error('Invalid Job, must have jobName set.')
		} else if(!(jobClass.prototype instanceof Job)) {
			throw new Error('All job classes must inherit from Job')
		}

		this.jobs[jobClass.jobName] = jobClass
	}

	findJobs(job) {
		if(typeof job === 'string') {
			job = this.jobs[job]
		}

		if(job.isNil || !(job.prototype instanceof Job)) {
			throw new Error('Invalid Job')
		}
	}

	query() {
		return new QueryBuilder(this)
	}

	fetchJob(id) {
		return new Promise((resolve, reject) => {
			Kue.Job.get(id, (err, job) => {
				if(!err.isNil) {
					return reject(err)
				}

				const jobClass = this.jobs[job.type]

				if(jobClass.isNil) {
					return reject(new Error(`Unable to fetch job of unknown type: ${job.type}`))
				}

				resolve(new jobClass(job.data, job))
			})
		})
	}

	dispatch(job) {
		return job.$save(this)
	}

	process(jobNames) {
		let jobs = Object.keys(this.jobs)

		if(!jobNames.isNil && jobs.length > 0) {
			jobs = jobs.filter(name => jobNames.indexOf(name) >= 0)
		}

		return Promise.all(jobs.map(
			key => this._process(this.jobs[key])
		))
	}

	_process(jobClass) {
		return new Promise(() => {
			this.kue.process(jobClass.jobName, jobClass.concurrency, async (kueJob, ctx, done) => {
				let result = null

				try {
					const job = new jobClass(kueJob.data, kueJob)
					result = job.$handle(this.app, this)
				} catch(err) {
					return done(this.handleError(err, done))
				}

				if(result.isNil || typeof result.then !== 'function') {
					return done()
				}

				result.then(() => done()).catch(err => this.handleError(err, done))
			})
		})
	}

	handleError(error, done) {
		return done(error)
	}

	destroy() {
		// TODO
	}

}<commit_msg>Added implementation for Queue.destroy()<commit_after>import './Job'
import './ConfigBuilder'
import './QueryBuilder'

import Kue from 'kue'

export class Queue {
	kue = null
	app = null
	jobs = { }

	constructor(app, config) {
		this.app = app
		this.kue = new Kue(ConfigBuilder(config, app))
	}

	register(jobClass) {
		if(jobClass.jobName.isNil) {
			throw new Error('Invalid Job, must have jobName set.')
		} else if(!(jobClass.prototype instanceof Job)) {
			throw new Error('All job classes must inherit from Job')
		}

		this.jobs[jobClass.jobName] = jobClass
	}

	findJobs(job) {
		if(typeof job === 'string') {
			job = this.jobs[job]
		}

		if(job.isNil || !(job.prototype instanceof Job)) {
			throw new Error('Invalid Job')
		}
	}

	query() {
		return new QueryBuilder(this)
	}

	fetchJob(id) {
		return new Promise((resolve, reject) => {
			Kue.Job.get(id, (err, job) => {
				if(!err.isNil) {
					return reject(err)
				}

				const jobClass = this.jobs[job.type]

				if(jobClass.isNil) {
					return reject(new Error(`Unable to fetch job of unknown type: ${job.type}`))
				}

				resolve(new jobClass(job.data, job))
			})
		})
	}

	dispatch(job) {
		return job.$save(this)
	}

	process(jobNames) {
		let jobs = Object.keys(this.jobs)

		if(!jobNames.isNil && jobs.length > 0) {
			jobs = jobs.filter(name => jobNames.indexOf(name) >= 0)
		}

		return Promise.all(jobs.map(
			key => this._process(this.jobs[key])
		))
	}

	_process(jobClass) {
		return new Promise(() => {
			this.kue.process(jobClass.jobName, jobClass.concurrency, async (kueJob, ctx, done) => {
				let result = null

				try {
					const job = new jobClass(kueJob.data, kueJob)
					result = job.$handle(this.app, this)
				} catch(err) {
					return done(this.handleError(err, done))
				}

				if(result.isNil || typeof result.then !== 'function') {
					return done()
				}

				result.then(() => done()).catch(err => this.handleError(err, done))
			})
		})
	}

	handleError(error, done) {
		return done(error)
	}

	destroy() {
		return new Promise((resolve, reject) => {
			this.kue.shutdown(5000, err => {
				if(!err.isNil) {
					return reject(err)
				}

				resolve()
			})
		})
	}

}
{'fraction_non_alphanumeric': 0.15966587112171837, 'fraction_numerical': 0.0019093078758949881, 'mean_word_length': 3.0531914893617023, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1143182', 'n_tokens_mistral': 1789, 'n_tokens_neox': 1497, 'n_words': 326}