bull

import Queue from 'bull'; import IORedis from 'ioredis'; // Ensure a Redis server is running at localhost:6379 or configure as needed. // For production, always use robust Redis connection settings. const REDIS_URL = process.env.REDIS_URL ?? 'redis://127.0.0.1:6379'; const connection = new IORedis(REDIS_URL); // Create a new queue named 'email-queue' const emailQueue = new Queue('email-queue', { connection }); console.log('Queue created. Adding a job and setting up a worker...'); // Producer: Add a job to the queue emailQueue.add('sendEmail', { to: 'user@example.com', subject: 'Welcome to Our Service', body: 'Hello there! Thanks for signing up.', }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 }, delay: 5000, // Delay job processing by 5 seconds removeOnComplete: true, // Automatically remove job from queue on completion }).then(job => { console.log(`Job '${job.id}' added to queue: sendEmail to ${job.data.to}`); }); // Consumer/Worker: Process jobs from the 'email-queue' emailQueue.process('sendEmail', async (job) => { const { to, subject, body } = job.data; console.log(`Processing job ${job.id}: Sending email to ${to} with subject '${subject}'`); // Simulate an asynchronous email sending operation await new Promise(resolve => setTimeout(resolve, Math.random() * 2000 + 500)); if (Math.random() < 0.1) { // 10% chance of failure console.error(`Job ${job.id} failed for ${to}. Will retry.`); throw new Error('Failed to send email (simulated error)'); } console.log(`Job ${job.id} completed: Email sent to ${to}.`); return { status: 'sent', recipient: to }; }); // Listen for global queue events emailQueue.on('completed', (job, result) => { console.log(`Global Event: Job ${job.id} completed with result:`, result); }); emailQueue.on('failed', (job, err) => { console.error(`Global Event: Job ${job.id} failed with error: ${err.message}`); }); console.log('Worker is listening for jobs...'); // Graceful shutdown process.on('SIGINT', async () => { console.log('Shutting down queues gracefully...'); await emailQueue.close(); await connection.quit(); console.log('Queues and Redis connection closed. Exiting.'); process.exit(0); });