How to use bull - 10 common examples
To help you get started, we’ve selected a few bull examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
flow-typed / flow-typed / definitions / npm / bull_v1.x.x / test_bull_v1.x.x.js View on Github 
'localhost',
);
Queue(
'my_queue',
9999,
'localhost',
);
new Queue(
'my_queue',
4444,
'localhost',
{ redisOption: true },
).process((job) => {
return Promise.resolve();
});
Queue(
'my_queue',
4444,
'localhost',
{ redisOption: true },
).process((job, done) => {
done();
});
// $ExpectError
Queue();
// $ExpectError
Queue(1);
Queue('my_queue', 9999, 'localhost')
.process((job, done) => {
Promise.resolve();
});constructor(options: TaskOptions) {
this.queue = new Queue(options.jobName, options.queue);
this.options = options;
// Sets up and attaches the job processor to the queue.
this.setupAndAttachProcessor();
}constructor({
jobName,
jobProcessor,
jobOptions = {},
queue,
}: TaskOptions) {
this.log = logger.child({ jobName }, true);
this.queue = new Queue(jobName, queue);
this.options = {
jobName,
jobProcessor,
jobOptions: {
// We always remove the job when it's complete, no need to fill up Redis
// with completed entries if we don't need to.
removeOnComplete: true,
// By default, configure jobs to use an exponential backoff
// strategy starting at a 10 second delay.
backoff: {
type: "exponential",
delay: 10000,
},
// Be default, try all jobs at least 5 times.};
export type IntegrationEvent = {
name: 'integrations.create' | 'integrations.update' | 'collections.delete',
modelId: string,
teamId: string,
actorId: string,
};
export type Event =
| UserEvent
| DocumentEvent
| CollectionEvent
| IntegrationEvent;
const globalEventsQueue = new Queue('global events', process.env.REDIS_URL);
const serviceEventsQueue = new Queue('service events', process.env.REDIS_URL);
// this queue processes global events and hands them off to service hooks
globalEventsQueue.process(async job => {
const names = Object.keys(services);
names.forEach(name => {
const service = services[name];
if (service.on) {
serviceEventsQueue.add(
{ service: name, ...job.data },
{ removeOnComplete: true }
);
}
});
});if (process.env.SMTP_USERNAME) {
smtpConfig.auth = {
user: process.env.SMTP_USERNAME,
pass: process.env.SMTP_PASSWORD,
};
}
this.transporter = nodemailer.createTransport(smtpConfig);
}
}
}
const mailer = new Mailer();
export default mailer;
export const mailerQueue = new Queue('email', process.env.REDIS_URL);
mailerQueue.process(async (job: EmailJob) => {
// $FlowIssue flow doesn't like dynamic values
await mailer[job.data.type](job.data.opts);
});
export const sendEmail = (type: Emails, to: string, options?: Object = {}) => {
mailerQueue.add(
{
type,
opts: {
to,
...options,
},
},
{import Queue from 'bull';
new Queue(
'my_queue',
9999,
'localhost',
);
Queue(
'my_queue',
9999,
'localhost',
);
new Queue(
'my_queue',
4444,
'localhost',
{ redisOption: true },
).process((job) => {
return Promise.resolve();
});
Queue(
'my_queue',
4444,
'localhost',
{ redisOption: true },
).process((job, done) => {
done();
});import { Throw } from '../utils/errors';
import Queue from 'bull';
import Arena from 'bull-arena';
import logger from '../utils/logger';
let version;
if (process.env.DOCKER) {
version = 'DOCKER';
} else {
version = require('../../../app/package.json').version;
}
const rssQueue = new Queue('rss', config.cache.uri);
const ogQueue = new Queue('og', config.cache.uri);
const podcastQueue = new Queue('podcast', config.cache.uri);
const socialQueue = new Queue('socail', config.cache.uri);
const streamQueue = new Queue('stream', config.cache.uri);
const tooOld = 3 * 60 * 60 * 1000;
const queues = {
'RSS Queue': rssQueue,
'OG Queue': ogQueue,
'Podcast Queue': podcastQueue,
'Social Score Queue': socialQueue,
'Personalisation-sync Queue': streamQueue,
};
exports.health = async (req, res) => {
res.status(200).send({ version, healthy: '100%' });
};import Podcast from '../models/podcast';
import moment from 'moment';
import config from '../config';
import { Throw } from '../utils/errors';
import Queue from 'bull';
import Arena from 'bull-arena';
import logger from '../utils/logger';
let version;
if (process.env.DOCKER) {
version = 'DOCKER';
} else {
version = require('../../../app/package.json').version;
}
const rssQueue = new Queue('rss', config.cache.uri);
const ogQueue = new Queue('og', config.cache.uri);
const podcastQueue = new Queue('podcast', config.cache.uri);
const socialQueue = new Queue('socail', config.cache.uri);
const streamQueue = new Queue('stream', config.cache.uri);
const tooOld = 3 * 60 * 60 * 1000;
const queues = {
'RSS Queue': rssQueue,
'OG Queue': ogQueue,
'Podcast Queue': podcastQueue,
'Social Score Queue': socialQueue,
'Personalisation-sync Queue': streamQueue,
};
exports.health = async (req, res) => {import moment from 'moment';
import config from '../config';
import { Throw } from '../utils/errors';
import Queue from 'bull';
import Arena from 'bull-arena';
import logger from '../utils/logger';
let version;
if (process.env.DOCKER) {
version = 'DOCKER';
} else {
version = require('../../../app/package.json').version;
}
const rssQueue = new Queue('rss', config.cache.uri);
const ogQueue = new Queue('og', config.cache.uri);
const podcastQueue = new Queue('podcast', config.cache.uri);
const socialQueue = new Queue('socail', config.cache.uri);
const streamQueue = new Queue('stream', config.cache.uri);
const tooOld = 3 * 60 * 60 * 1000;
const queues = {
'RSS Queue': rssQueue,
'OG Queue': ogQueue,
'Podcast Queue': podcastQueue,
'Social Score Queue': socialQueue,
'Personalisation-sync Queue': streamQueue,
};
exports.health = async (req, res) => {
res.status(200).send({ version, healthy: '100%' });import Queue from 'bull';
import Arena from 'bull-arena';
import logger from '../utils/logger';
let version;
if (process.env.DOCKER) {
version = 'DOCKER';
} else {
version = require('../../../app/package.json').version;
}
const rssQueue = new Queue('rss', config.cache.uri);
const ogQueue = new Queue('og', config.cache.uri);
const podcastQueue = new Queue('podcast', config.cache.uri);
const socialQueue = new Queue('socail', config.cache.uri);
const streamQueue = new Queue('stream', config.cache.uri);
const tooOld = 3 * 60 * 60 * 1000;
const queues = {
'RSS Queue': rssQueue,
'OG Queue': ogQueue,
'Podcast Queue': podcastQueue,
'Social Score Queue': socialQueue,
'Personalisation-sync Queue': streamQueue,
};
exports.health = async (req, res) => {
res.status(200).send({ version, healthy: '100%' });
};
exports.status = async (req, res) => {
