index.js
/**
 * Simple SQLite backed Queue for running many short tasks in Node.js
 *
 * @author Damien Clark <damo.clarky@gmail.com>
 * @param {string} [filename=:memory:] Path to sqlite db for queue db
 * @param {number} [batchSize=10] The number of rows from queue db to retrieve at a time
 * @constructor
 */
Simple SQLite backed Queue for running many short tasks in Node.js
@author Damien Clark 
@param {string} [filename=:memory:] Path to sqlite db for queue db
@param {number} [batchSize=10] The number of rows from queue db to retrieve at a time
@constructor
Simple SQLite backed Queue for running many short tasks in Node . js @author Damien Clark @param { string } [ filename = : memory : ] Path to sqlite db for queue db @param { number } [ batchSize = 10 ] The number of rows from queue db to retrieve at a time @constructor
function PersistentQueue(filename, batchSize) {
	EventEmitter.call(this) ;
	if(filename === undefined)
		throw new Error('No filename parameter provided') ;
	this.debug = false ;
	this.empty = undefined ;
	this.dbPath = (filename === '') ? ':memory:' : filename ;
	this.batchSize = (batchSize === undefined) ? 10 : batchSize ;
	if(typeof this.batchSize !== 'number' || this.batchSize < 1)
		throw new Error('Invalid batchSize parameter.  Must be a number > 0') ;
	this.queue = [] ;
	this.length  = null ;
	this.db = null ;
	this.opened = false ;
	this.run = false ;
	this.on('start', () => {
		if(this.db === null)
			throw new Error('Open queue database before starting queue') ;
		if(this.run === false) {
			this.run = true ;
			this.emit('trigger_next') ;
		}
	}) ;
	this.on('stop', () => {
		this.run = false ;
	}) ;
	this.on('trigger_next', () => {
		if(this.debug) console.log('trigger_next') ;
		if(!this.run || this.empty) {
			if(this.debug) console.log('run='+this.run+' and empty='+this.empty) ;
			if(this.debug) console.log('not started or empty queue') ;
			return ;
		}
		const trigger = () => {
			this.emit('next', this.queue[0]) ;
		} ;
		if(this.queue.length === 0 && this.length !== 0) {
			hydrateQueue(this, this.batchSize)
			.then(() => {
				setImmediate(trigger) ;
			})
			.catch(err => {
				console.error(err) ;
				process.exit(1) ;
			}) ;
		}
		else if(this.queue.length) { 
			setImmediate(trigger) ;
		}
		else { 
			this.emit('empty') ;
		}
	}) ;
	this.on('empty', () => {
		this.empty = true ;
		this.db.exec("VACUUM;")
	}) ;
	this.on('add', job => {
		if(this.empty) {
			this.empty = false ;
			if(this.debug) console.log('No longer empty') ;
			if(this.run)
				this.emit('trigger_next') ;
		}
	}) ;
	this.on('open', db => {
		this.opened = true ;
	}) ;
	this.on('close', () => {
		this.opened = false ;
		this.db = null ;
		this.empty = undefined ;
		this.run = false ;
		this.queue = [] ;
	}) ;
}
function PersistentQueue ( filename , batchSize ) { EventEmitter . call ( this ) ; if ( filename === undefined ) throw new Error ( 'No filename parameter provided' ) ; this . debug = false ; this . empty = undefined ; this . dbPath = ( filename === '' ) ? ':memory:' : filename ; this . batchSize = ( batchSize === undefined ) ? 10 : batchSize ; if ( typeof this . batchSize !== 'number' || this . batchSize < 1 ) throw new Error ( 'Invalid batchSize parameter.  Must be a number > 0' ) ; this . queue = [ ] ; this . length = null ; this . db = null ; this . opened = false ; this . run = false ; this . on ( 'start' , ( ) => { if ( this . db === null ) throw new Error ( 'Open queue database before starting queue' ) ; if ( this . run === false ) { this . run = true ; this . emit ( 'trigger_next' ) ; } } ) ; this . on ( 'stop' , ( ) => { this . run = false ; } ) ; this . on ( 'trigger_next' , ( ) => { if ( this . debug ) console . log ( 'trigger_next' ) ; if ( ! this . run || this . empty ) { if ( this . debug ) console . log ( 'run=' + this . run + ' and empty=' + this . empty ) ; if ( this . debug ) console . log ( 'not started or empty queue' ) ; return ; } const trigger = ( ) => { this . emit ( 'next' , this . queue [ 0 ] ) ; } ; if ( this . queue . length === 0 && this . length !== 0 ) { hydrateQueue ( this , this . batchSize ) . then ( ( ) => { setImmediate ( trigger ) ; } ) . catch ( err => { console . error ( err ) ; process . exit ( 1 ) ; } ) ; } else if ( this . queue . length ) { setImmediate ( trigger ) ; } else { this . emit ( 'empty' ) ; } } ) ; this . on ( 'empty' , ( ) => { this . empty = true ; this . db . exec ( "VACUUM;" ) } ) ; this . on ( 'add' , job => { if ( this . empty ) { this . empty = false ; if ( this . debug ) console . log ( 'No longer empty' ) ; if ( this . run ) this . emit ( 'trigger_next' ) ; } } ) ; this . on ( 'open' , db => { this . opened = true ; } ) ; this . on ( 'close' , ( ) => { this . opened = false ; this . db = null ; this . empty = undefined ; this . run = false ; this . queue = [ ] ; } ) ; }
Simple SQLite backed Queue for running many short tasks in Node.js
 @author Damien Clark <damo.clarky@gmail.com>
 @param {string} [filename=:memory:] Path to sqlite db for queue db
 @param {number} [batchSize=10] The number of rows from queue db to retrieve at a time
 @constructor
Simple SQLite backed Queue for running many short tasks in Node . js @author Damien Clark <damo . clarky@gmail . com > @param { string } [ filename = : memory : ] Path to sqlite db for queue db @param { number } [ batchSize = 10 ] The number of rows from queue db to retrieve at a time @constructor
// Call super-constructor // If filename not provided, then throw error /**
	 * Set to true to enable debugging mode
	 * @type {boolean}
	 * @access private
	 */ /**
	 * Instance variable for whether the queue is empty (not known at instantiation)
	 * @type {boolean}
	 * @access private
	 */ /**
	 * Path to the sqlite db file
	 * @type {string}
	 * @access private
	 */ /**
	 * How many objects to retrieve from DB into queue array at a time
	 */ /**
	 * The queue of objects to operate on
	 * @type {Array}
	 * @access private
	 */ /**
	 * Keep track of total number of jobs in queue
	 * @type {number}
	 * @access private
	 */ /**
	 * The sqlite database object handle
	 * @type {sqlite3.Database}
	 * @access private
	 */ /**
	 * The queue's sqlite database is open
	 * @type {boolean}
	 * @access private
	 */ /**
	 * Should the queue process messages
	 * @type {boolean}
	 * @access private
	 */ //Check state of queue // If queue not started or is empty, then just return // Define our embedded recursive function to be called later // If our in-memory list is empty, but queue is not, re-hydrate from db // Schedule job for next check phase in event loop // If in-memory queue not empty, trigger next job // https://nodejs.org/en/docs/guides/event-loop-timers-and-nexttick/ // Otherwise queue is empty // Set instance to empty on empty event // Ask sqlite to free up unused space // If a job is added, trigger_next event // eslint-disable-next-line no-unused-vars // eslint-disable-next-line no-unused-vars // Unset the db variable when db is closed