#257

Build # #257

Build Type

push

Committed by

Commit Message

<a href="https://github.com/taskforcesh/bullmq/commit/<a class=hub.com/taskforcesh/bullmq/commit/bb7839228c5950633cc39ec706084c3e3c8afa17">bb7839228<a href="https://github.com/taskforcesh/bullmq/commit/bb7839228c5950633cc39ec706084c3e3c8afa17">">Merge </a><a class="double-link" href="https://github.com/taskforcesh/bullmq/commit/<a class="double-link" href="https://github.com/taskforcesh/bullmq/commit/f0de37f0352524a578cce35d0996e5e612ffe542">f0de37f03</a>">f0de37f03</a><a href="https://github.com/taskforcesh/bullmq/commit/bb7839228c5950633cc39ec706084c3e3c8afa17"> into de99a07c3">de99a07c3</a>

Run Details

616 of 841 branches covered (73.25%)

Branch coverage included in aggregate %.

4080 of 4506 relevant lines covered (90.55%)

74.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.2

/src/test/test_delay.ts

import { Queue, Job } from '@src/classes';
import { describe, beforeEach, it } from 'mocha';
import { expect } from 'chai';
import * as IORedis from 'ioredis';
import { v4 } from 'uuid';
import { Worker } from '@src/classes/worker';
import { QueueEvents } from '@src/classes/queue-events';
import { QueueScheduler } from '@src/classes/queue-scheduler';
import { removeAllQueueData } from '@src/utils';

describe('Delayed jobs', function() {
  this.timeout(15000);

  let queue: Queue;
  let queueName: string;

  beforeEach(async function() {
    queueName = 'test-' + v4();
    queue = new Queue(queueName);
  });

  afterEach(async function() {
    await queue.close();
    await removeAllQueueData(new IORedis(), queueName);
  });

  it('should process a delayed job only after delayed time', async function() {
    const delay = 1000;
    const queueScheduler = new QueueScheduler(queueName);
    await queueScheduler.waitUntilReady();
    const queueEvents = new QueueEvents(queueName);
    await queueEvents.waitUntilReady();

    const worker = new Worker(queueName, async job => {});

    const timestamp = Date.now();
    let publishHappened = false;

    queueEvents.on('delayed', () => {
      publishHappened = true;
    });

    const completed = new Promise((resolve, reject) => {
      queueEvents.on('completed', async function() {
        try {
          expect(Date.now() > timestamp + delay);
          const jobs = await queue.getWaiting();
          expect(jobs.length).to.be.equal(0);

          const delayedJobs = await queue.getDelayed();
          expect(delayedJobs.length).to.be.equal(0);
          expect(publishHappened).to.be.eql(true);
          await worker.close();
          resolve();
        } catch (err) {
          reject(err);
        }
      });
    });

    const job = await queue.add('test', { delayed: 'foobar' }, { delay });

    expect(job.id).to.be.ok;
    expect(job.data.delayed).to.be.eql('foobar');
    expect(job.opts.delay).to.be.eql(delay);

    await completed;
    await queueScheduler.close();
    await queueEvents.close();
  });

  it('should process delayed jobs in correct order', async function() {
    this.timeout(20000);
    let order = 0;
    const queueScheduler = new QueueScheduler(queueName);
    await queueScheduler.waitUntilReady();

    const processing = new Promise((resolve, reject) => {
      const processor = async (job: Job) => {
        order++;
        try {
          expect(order).to.be.equal(job.data.order);
          if (order === 10) {
            resolve(worker.close());
          }
        } catch (err) {
          reject(err);
        }
      };

      const worker = new Worker(queueName, processor);

      worker.on('failed', function(job, err) {
        err.job = job;
      });
    });

    await Promise.all([
      queue.add('test', { order: 1 }, { delay: 100 }),
      queue.add('test', { order: 6 }, { delay: 600 }),
      queue.add('test', { order: 10 }, { delay: 1000 }),
      queue.add('test', { order: 2 }, { delay: 200 }),
      queue.add('test', { order: 9 }, { delay: 900 }),
      queue.add('test', { order: 5 }, { delay: 500 }),
      queue.add('test', { order: 3 }, { delay: 300 }),
      queue.add('test', { order: 7 }, { delay: 700 }),
      queue.add('test', { order: 4 }, { delay: 400 }),
      queue.add('test', { order: 8 }, { delay: 800 }),
    ]);

    await processing;

    await queueScheduler.close();
  });

  it('should process delayed jobs in correct order even in case of restart', async function() {
    this.timeout(5000);

    let worker: Worker;
    const queueName = 'delayed queue multiple' + v4();
    let order = 1;

    let secondQueueScheduler: QueueScheduler;
    const firstQueueScheduler = new QueueScheduler(queueName);
    await firstQueueScheduler.waitUntilReady();

    queue = new Queue(queueName);

    const processing = new Promise((resolve, reject) => {
      worker = new Worker(queueName, async (job: Job) => {
        try {
          expect(order).to.be.equal(job.data.order);

          if (order === 1) {
            await firstQueueScheduler.close();
            secondQueueScheduler = new QueueScheduler(queueName);
            await secondQueueScheduler.waitUntilReady();
          }

          if (order === 4) {
            resolve();
          }
          order++;
        } catch (err) {
          reject(err);
        }
      });
    });

    await Promise.all([
      queue.add('test', { order: 2 }, { delay: 500 }),
      queue.add('test', { order: 4 }, { delay: 1500 }),
      queue.add('test', { order: 1 }, { delay: 200 }),
      queue.add('test', { order: 3 }, { delay: 800 }),
    ]);

    await processing;

    await queue.close();
    worker && (await worker.close());
    secondQueueScheduler && (await secondQueueScheduler.close());
  });

  it('should process delayed jobs with exact same timestamps in correct order (FIFO)', async function() {
    let order = 1;

    const queueScheduler = new QueueScheduler(queueName);
    await queueScheduler.waitUntilReady();

    const processing = new Promise((resolve, reject) => {
      const processor = async (job: Job) => {
        try {
          expect(order).to.be.equal(job.data.order);

          if (order === 12) {
            resolve(worker.close());
          }
        } catch (err) {
          reject(err);
        }

        order++;
      };

      const worker = new Worker(queueName, processor);

      worker.on('failed', function(job, err) {
        err.job = job;
      });
    });

    const now = Date.now();
    const promises = [];
    let i = 1;
    for (i; i <= 12; i++) {
      promises.push(
        queue.add(
          'test',
          { order: i },
          {
            delay: 1000,
            timestamp: now,
          },
        ),
      );
    }
    await Promise.all(promises);
    await processing;
    await queueScheduler.close();
  });
});

1	import { Queue, Job } from '@src/classes';	1✔
2	import { describe, beforeEach, it } from 'mocha';	1✔
3	import { expect } from 'chai';	1✔
4	import * as IORedis from 'ioredis';	1✔
5	import { v4 } from 'uuid';	1✔
6	import { Worker } from '@src/classes/worker';	1✔
7	import { QueueEvents } from '@src/classes/queue-events';	1✔
8	import { QueueScheduler } from '@src/classes/queue-scheduler';	1✔
9	import { removeAllQueueData } from '@src/utils';	1✔
10
11	describe('Delayed jobs', function() {	1✔
12	this.timeout(15000);	1✔
13
14	let queue: Queue;
15	let queueName: string;
16
17	beforeEach(async function() {	1✔
18	queueName = 'test-' + v4();	4✔
19	queue = new Queue(queueName);	4✔
20	});
21
22	afterEach(async function() {	1✔
23	await queue.close();	4✔
24	await removeAllQueueData(new IORedis(), queueName);	4✔
25	});
26
27	it('should process a delayed job only after delayed time', async function() {	1✔
28	const delay = 1000;	1✔
29	const queueScheduler = new QueueScheduler(queueName);	1✔
30	await queueScheduler.waitUntilReady();	1✔
31	const queueEvents = new QueueEvents(queueName);	1✔
32	await queueEvents.waitUntilReady();	1✔
33
34	const worker = new Worker(queueName, async job => {});	1✔
35
36	const timestamp = Date.now();	1✔
37	let publishHappened = false;	1✔
38
39	queueEvents.on('delayed', () => {	1✔
40	publishHappened = true;	1✔
41	});
42
43	const completed = new Promise((resolve, reject) => {	1✔
44	queueEvents.on('completed', async function() {	1✔
45	try {	1✔
46	expect(Date.now() > timestamp + delay);	1✔
47	const jobs = await queue.getWaiting();	1✔
48	expect(jobs.length).to.be.equal(0);	1✔
49
50	const delayedJobs = await queue.getDelayed();	1✔
51	expect(delayedJobs.length).to.be.equal(0);	1✔
52	expect(publishHappened).to.be.eql(true);	1✔
53	await worker.close();	1✔
54	resolve();	1✔
55	} catch (err) {
56	reject(err);	×
57	}
58	});
59	});
60
61	const job = await queue.add('test', { delayed: 'foobar' }, { delay });	1✔
62
63	expect(job.id).to.be.ok;	1✔
64	expect(job.data.delayed).to.be.eql('foobar');	1✔
65	expect(job.opts.delay).to.be.eql(delay);	1✔
66
67	await completed;	1✔
68	await queueScheduler.close();	1✔
69	await queueEvents.close();	1✔
70	});
71
72	it('should process delayed jobs in correct order', async function() {	1✔
73	this.timeout(20000);	1✔
74	let order = 0;	1✔
75	const queueScheduler = new QueueScheduler(queueName);	1✔
76	await queueScheduler.waitUntilReady();	1✔
77
78	const processing = new Promise((resolve, reject) => {	1✔
79	const processor = async (job: Job) => {	1✔
80	order++;	10✔
81	try {	10✔
82	expect(order).to.be.equal(job.data.order);	10✔
83	if (order === 10) {	10✔
84	resolve(worker.close());	1✔
85	}
86	} catch (err) {
87	reject(err);	×
88	}
89	};
90
91	const worker = new Worker(queueName, processor);	1✔
92
93	worker.on('failed', function(job, err) {	1✔
94	err.job = job;	×
95	});
96	});
97
98	await Promise.all([	1✔
99	queue.add('test', { order: 1 }, { delay: 100 }),
100	queue.add('test', { order: 6 }, { delay: 600 }),
101	queue.add('test', { order: 10 }, { delay: 1000 }),
102	queue.add('test', { order: 2 }, { delay: 200 }),
103	queue.add('test', { order: 9 }, { delay: 900 }),
104	queue.add('test', { order: 5 }, { delay: 500 }),
105	queue.add('test', { order: 3 }, { delay: 300 }),
106	queue.add('test', { order: 7 }, { delay: 700 }),
107	queue.add('test', { order: 4 }, { delay: 400 }),
108	queue.add('test', { order: 8 }, { delay: 800 }),
109	]);
110
111	await processing;	1✔
112
113	await queueScheduler.close();	1✔
114	});
115
116	it('should process delayed jobs in correct order even in case of restart', async function() {	1✔
117	this.timeout(5000);	1✔
118
119	let worker: Worker;
120	const queueName = 'delayed queue multiple' + v4();	1✔
121	let order = 1;	1✔
122
123	let secondQueueScheduler: QueueScheduler;
124	const firstQueueScheduler = new QueueScheduler(queueName);	1✔
125	await firstQueueScheduler.waitUntilReady();	1✔
126
127	queue = new Queue(queueName);	1✔
128
129	const processing = new Promise((resolve, reject) => {	1✔
130	worker = new Worker(queueName, async (job: Job) => {	1✔
131	try {	4✔
132	expect(order).to.be.equal(job.data.order);	4✔
133
134	if (order === 1) {	4✔
135	await firstQueueScheduler.close();	1✔
136	secondQueueScheduler = new QueueScheduler(queueName);	1✔
137	await secondQueueScheduler.waitUntilReady();	1✔
138	}
139
140	if (order === 4) {	4✔
141	resolve();	1✔
142	}
143	order++;	4✔
144	} catch (err) {
145	reject(err);	×
146	}
147	});
148	});
149
150	await Promise.all([	1✔
151	queue.add('test', { order: 2 }, { delay: 500 }),
152	queue.add('test', { order: 4 }, { delay: 1500 }),
153	queue.add('test', { order: 1 }, { delay: 200 }),
154	queue.add('test', { order: 3 }, { delay: 800 }),
155	]);
156
157	await processing;	1✔
158
159	await queue.close();	1✔
160	worker && (await worker.close());	1✔
161	secondQueueScheduler && (await secondQueueScheduler.close());	1✔
162	});
163
164	it('should process delayed jobs with exact same timestamps in correct order (FIFO)', async function() {	1✔
165	let order = 1;	1✔
166
167	const queueScheduler = new QueueScheduler(queueName);	1✔
168	await queueScheduler.waitUntilReady();	1✔
169
170	const processing = new Promise((resolve, reject) => {	1✔
171	const processor = async (job: Job) => {	1✔
172	try {	12✔
173	expect(order).to.be.equal(job.data.order);	12✔
174
175	if (order === 12) {	12✔
176	resolve(worker.close());	1✔
177	}
178	} catch (err) {
179	reject(err);	×
180	}
181
182	order++;	12✔
183	};
184
185	const worker = new Worker(queueName, processor);	1✔
186
187	worker.on('failed', function(job, err) {	1✔
188	err.job = job;	×
189	});
190	});
191
192	const now = Date.now();	1✔
193	const promises = [];	1✔
194	let i = 1;	1✔
195	for (i; i <= 12; i++) {	1✔
196	promises.push(	12✔
197	queue.add(
198	'test',
199	{ order: i },
200	{
201	delay: 1000,
202	timestamp: now,
203	},
204	),
205	);
206	}
207	await Promise.all(promises);	1✔
208	await processing;	1✔
209	await queueScheduler.close();	1✔
210	});
211	});

taskforcesh / bullmq / #257

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous