-
Notifications
You must be signed in to change notification settings - Fork 155
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Feature] Add queue and queue message classes
- Loading branch information
Showing
6 changed files
with
310 additions
and
90 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| const PgBoss = require("pg-boss"); | ||
| const PgBossKnexAdapter = require("./pg_boss_knex_adapter"); | ||
| const util = require("util"); | ||
|
|
||
| /** | ||
| * Implements a message queue using the PgBoss library. | ||
| */ | ||
| class Queue { | ||
| #queueClient; | ||
| #queueName; | ||
| #messageClass; | ||
| #logger; | ||
|
|
||
| /** | ||
| * @param {object} params the parameter object | ||
| * @param {import('pg-boss')} params.queueClient the queue client instance to | ||
| * use for queue operations. | ||
| * @param {string} params.queueName the identifier for the queue. | ||
| * @param {*} params.messageClass a class which implements the fromMessage | ||
| * static method to return an instance of the class from a PgBoss message | ||
| * object. This can be omitted if the queue instance only sends messages. | ||
| * @param {import('winston').Logger} params.logger an application logger instance. | ||
| */ | ||
| constructor({ queueClient, queueName, messageClass, logger }) { | ||
| this.#queueClient = queueClient; | ||
| this.#queueName = queueName; | ||
| this.#messageClass = messageClass; | ||
| this.#logger = logger; | ||
| } | ||
|
|
||
| /** | ||
| * @returns {string} the queue name | ||
| */ | ||
| get name() { | ||
| return this.#queueName; | ||
| } | ||
|
|
||
| /** | ||
| * @returns {Promise} resolves when the PgBoss queue client has started | ||
| */ | ||
| async start() { | ||
| try { | ||
| await this.#queueClient.start(); | ||
| this.#logger.debug("Starting queue client"); | ||
| } catch (e) { | ||
| this.#logger.error("Error starting queue client"); | ||
| this.#logger.error(util.inspect(e)); | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @returns {Promise} resolves when the PgBoss queue client has stopped | ||
| */ | ||
| async stop() { | ||
| try { | ||
| await this.#queueClient.stop(); | ||
| this.#logger.debug(`Stopping queue client`); | ||
| } catch (e) { | ||
| this.#logger.error("Error stopping queue client"); | ||
| this.#logger.error(util.inspect(e)); | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @param {import('./queue_message')} queueMessage a QueueMessage instance | ||
| * @returns {string} a message ID or null if a duplicate message exists on the | ||
| * queue. | ||
| */ | ||
| async sendMessage(queueMessage) { | ||
| try { | ||
| const messageId = await this.#queueClient.send( | ||
| this.#queueName, | ||
| queueMessage.toJSON(), | ||
| queueMessage.sendOptions(), | ||
| ); | ||
| if (messageId) { | ||
| this.#logger.info( | ||
| `Created message in queue: ${this.#queueName} with message ID: ${messageId}`, | ||
| ); | ||
| } else { | ||
| this.#logger.info( | ||
| `Found a duplicate message in queue: ${this.#queueName}`, | ||
| ); | ||
| } | ||
| return messageId; | ||
| } catch (e) { | ||
| this.#logger.error(`Error sending to queue: ${this.#queueName}`); | ||
| this.#logger.error(util.inspect(e)); | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @param {Function} callback the function to call for each message | ||
| * @param {object} options the options to pass to the PgBoss work function | ||
| * @returns {Promise} resolves when the queue poller process stops | ||
| */ | ||
| poll(callback, options = { newJobCheckIntervalSeconds: 1 }) { | ||
| return this.#queueClient.work(this.#queueName, options, async (message) => { | ||
| this.#logger.info("Queue message received"); | ||
| await callback(this.#messageClass.fromMessage(message).toJSON()); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * @param {object} params the parameter object | ||
| * @param {import('knex')} params.knexInstance an initialized instance of the knex | ||
| * library which provides a database connection. | ||
| * @param {string} params.queueName the name of the queue to use for the | ||
| * client. | ||
| * @param {*} params.messageClass a class which implements the fromMessage | ||
| * static method to return an instance of the class from a PgBoss message | ||
| * object. This can be omitted if the queue instance only sends messages. | ||
| * @param {import('winston').Logger} params.logger an application logger instance. | ||
| * @returns {Queue} the queue instance configured with the PgBoss queue | ||
| * client. | ||
| */ | ||
| static buildQueue({ knexInstance, queueName, messageClass, logger }) { | ||
| return new Queue({ | ||
| queueClient: new PgBoss({ db: new PgBossKnexAdapter(knexInstance) }), | ||
| queueName, | ||
| messageClass, | ||
| logger, | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| module.exports = Queue; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| /** | ||
| * Abstract class for a queue message to be sent to a PgBoss queue client. | ||
| */ | ||
| class QueueMessage { | ||
| /** | ||
| * @returns {object} the class converted to a JSON object. | ||
| */ | ||
| toJSON() { | ||
| return {}; | ||
| } | ||
|
|
||
| /** | ||
| * @returns {object} an options object for the PgBoss send method | ||
| */ | ||
| sendOptions() { | ||
| return {}; | ||
| } | ||
|
|
||
| /** | ||
| * @param {object} message a PgBoss message object from the report job queue. | ||
| * @returns {QueueMessage} the built queue message instance. | ||
| */ | ||
| static fromMessage(message) { | ||
| return new QueueMessage(message.data); | ||
| } | ||
| } | ||
|
|
||
| module.exports = QueueMessage; |
Oops, something went wrong.