groot/automatisch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #2246 from automatisch/restructure-queues

Browse Source 
refactor: Abstract queue generation and configuration
This commit is contained in:
Ömer Faruk Aydın
2024-12-12 14:57:03 +01:00
committed by GitHub
parent 978ceaadb6 e41a331ad7
commit 3d62fabaac
7 changed files with 57 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
packages/backend/src/queues/action.js
View File

@@ -1,27 +1,4 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
const actionQueue = new Queue('action', redisConnection);
actionQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error('Error happened in action queue!', error);
});
import { generateQueue } from './queue.js';
const actionQueue = generateQueue('action');
export default actionQueue;

27
packages/backend/src/queues/delete-user.ee.js
View File

@@ -1,27 +1,4 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
const deleteUserQueue = new Queue('delete-user', redisConnection);
deleteUserQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error('Error happened in delete user queue!', error);
});
import { generateQueue } from './queue.js';
const deleteUserQueue = generateQueue('delete-user');
export default deleteUserQueue;

27
packages/backend/src/queues/email.js
View File

@@ -1,27 +1,4 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
const emailQueue = new Queue('email', redisConnection);
emailQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error('Error happened in email queue!', error);
});
import { generateQueue } from './queue.js';
const emailQueue = generateQueue('email');
export default emailQueue;

27
packages/backend/src/queues/flow.js
View File

@@ -1,27 +1,4 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
const flowQueue = new Queue('flow', redisConnection);
flowQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error('Error happened in flow queue!', error);
});
import { generateQueue } from './queue.js';
const flowQueue = generateQueue('flow');
export default flowQueue;

44
packages/backend/src/queues/queue.js Normal file
View File

@@ -0,0 +1,44 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
export const generateQueue = (queueName, options) => {
const queue = new Queue(queueName, redisConnection);
queue.on('error', (error) => queueOnError(error, queueName));
if (options?.runDaily) addScheduler(queueName, queue);
return queue;
};
const queueOnError = (error, queueName) => {
if (error.code === CONNECTION_REFUSED) {
const errorMessage =
'Make sure you have installed Redis and it is running.';
logger.error(errorMessage, error);
process.exit();
}
logger.error(`Error happened in ${queueName} queue!`, error);
};
const addScheduler = (queueName, queue) => {
const everydayAtOneOclock = '0 1 * * *';
queue.add(queueName, null, {
jobId: queueName,
repeat: {
pattern: everydayAtOneOclock,
},
});
};

38
packages/backend/src/queues/remove-cancelled-subscriptions.ee.js
View File

@@ -1,40 +1,8 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
import { generateQueue } from './queue.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
const removeCancelledSubscriptionsQueue = new Queue(
const removeCancelledSubscriptionsQueue = generateQueue(
'remove-cancelled-subscriptions',
redisConnection
{ runDaily: true }
);
removeCancelledSubscriptionsQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error(
'Error happened in remove cancelled subscriptions queue!',
error
);
});
removeCancelledSubscriptionsQueue.add('remove-cancelled-subscriptions', null, {
jobId: 'remove-cancelled-subscriptions',
repeat: {
pattern: '0 1 * * *',
},
});
export default removeCancelledSubscriptionsQueue;

27
packages/backend/src/queues/trigger.js
View File

@@ -1,27 +1,4 @@
import process from 'process';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
const redisConnection = {
connection: redisConfig,
};
const triggerQueue = new Queue('trigger', redisConnection);
triggerQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error('Error happened in trigger queue!', error);
});
import { generateQueue } from './queue.js';
const triggerQueue = generateQueue('trigger');
export default triggerQueue;