Task queue service with graphile-worker
npm install moleculer-psql-queueJob queue mixin for graphile-worker.
``
$ npm install moleculer-psql-queue
1. Start your PostgreSQL db.
2. Create an empty db psql -U postgres -c 'CREATE DATABASE task_queue'.
> Replace task_queue with your db name
3. Use graphile-worker CLI to init the schema for the jobs npx graphile-worker -c "postgres://postgres:postgres@localhost:5444/task_queue" --schema-only.
> Set your connection URL (more info: check PSQL docs) and replace task_queue with db name that you've defined in step 2)
`js
const PsqlQueueService = require("moleculer-psql-queue");
broker.createService({
name: "task-worker",
mixins: [
PsqlQueueService(
"postgres://postgres:postgres@localhost:5444/task_queue",
// Default opts
{
// Name of the property in service schema.
schemaProperty: "queues",
// Name of the method in Service to create jobs
createJobMethodName: "createJob",
// Name of the property in Service to produce jobs
producerPropertyName: "$producer",
// Name of the property in Service to consume jobs
consumerPropertyName: "$consumer",
// Name of the internal queue that's used to store the job handlers
internalQueueName: "$queue",
// Name of the property in Service settings to register job event handlers
jobEventHandlersSettingsProperty: "jobEventHandlers",
// Optional producer configs: More info: https://github.com/graphile/worker#workerutilsoptions
producerOpts: {},
// Optional worker configs. More info: https://github.com/graphile/worker#runneroptions
queueOpts: {
concurrency: 5,
// Install signal handlers for graceful shutdown on SIGINT, SIGTERM, etc
noHandleSignals: false,
pollInterval: 1000,
},
}
),
],
queues: {
/**
* @param {Object} payload Message payload
* @param {import('graphile-worker').JobHelpers} helpers graphile-worker
* More info about helpers: https://github.com/graphile/worker#creating-task-executors
*/
"sample.task"(payload, helpers) {
// if (Math.random() > 0.5) {
this.logger.info('New "simple.task" received!', payload);
return;
// } else {
// throw new Error('Random "sample.task" error');
// }
},
"another.task": {
/**
* @param {Object} payload Message payload
* @param {import('graphile-worker').JobHelpers} helpers Postgres helpers
* More info about helpers: https://github.com/graphile/worker#creating-task-executors
*/
process(payload, helpers) {
this.logger.info('New "another.task" job received!', payload);
},
},
},
});
`
`js
const PsqlQueueService = require("moleculer-psql-queue");
broker.createService({
name: "task-worker",
mixins: [
PsqlQueueService(
"postgres://postgres:postgres@localhost:5444/task_queue"
),
],
methods: {
/**
* Replaces Default logger with custom one.
* By default uses Moleculer logger instance
* More info: https://github.com/graphile/worker#logger
*/
initLogger() {
/**
* @param {String} level Log level
* @param {String} message Message to log
* @param {Object} meta Additional metadata
*/
return (level, message, meta) => {
this.loggerQueuelevel;
};
},
},
// Add Workers here
queues: {},
});
`
`js
const PsqlQueueService = require("moleculer-psql-queue");
broker.createService({
name: "task-worker",
mixins: [
PsqlQueueService(
"postgres://postgres:postgres@localhost:5444/task_queue"
),
],
settings: {
/**
* @type {Record
* For a complete list of events see: https://github.com/graphile/worker#workerevents
*/
jobEventHandlers: {
/**
* @param {{
* worker: import('graphile-worker').Worker,
* job: import('graphile-worker').Job
* }}
* @this {import('moleculer').Service}
*/
"job:success": function ({ worker, job }) {
this.logger.info(
Worker ${worker.workerId} completed job ${job.id}
);
},
},
},
// Add Workers here
queues: {},
});
`
`js
const PsqlQueueService = require("moleculer-psql-queue");
broker.createService({
name: "pub",
mixins: [
PsqlQueueService(
"postgres://postgres:postgres@localhost:5444/task_queue"
),
],
/**
* Service started lifecycle event handler
* @this {import('moleculer').Service}
*/
async started() {
try {
/**
* @param {String} name Task name
* @param {Object} payload Payload to pass to the task
* @param {import('graphile-worker').TaskSpec?} opts
*/
await this.createJob("sample.task", {
id: 1,
name: "simple.task",
});
} catch (error) {
this.logger.error('Error creating "sample.task" job', error);
}
},
});
`
The graphile-worker lib provides some advanced features like administration functions. These functions can be used to manage the queue and can be accessed via the this.$producer property of the service.
`js
const PsqlQueueService = require("moleculer-psql-queue");
broker.createService({
name: "pub",
mixins: [
PsqlQueueService(
"postgres://postgres:postgres@localhost:5444/task_queue"
),
],
/**
* Service started lifecycle event handler
* @this {import('moleculer').Service}
*/
async started() {
// Add the job via raw graphile-worker client
// For more info check the docs: https://github.com/graphile/worker#administration-functions
this.$producer.addJob("sample.task", {
id: 1,
name: "simple.task",
});
},
});
`
`
$ npm test
In development with watching
`
$ npm run ci
The project is available under the MIT license.
Copyright (c) 2016-2022 MoleculerJS
 