#269

Build # #269

Build Type

push

Committed by

Commit Message

<a href="https://github.com/taskforcesh/bullmq/commit/<a class=hub.com/taskforcesh/bullmq/commit/9acb18ce537a55a2fb60350ff216d15140aad68a">9acb18ce5<a href="https://github.com/taskforcesh/bullmq/commit/9acb18ce537a55a2fb60350ff216d15140aad68a">">Merge </a><a class="double-link" href="https://github.com/taskforcesh/bullmq/commit/<a class="double-link" href="https://github.com/taskforcesh/bullmq/commit/ef7e7f67d9c587ffdf4c56f34500b65a527e6ba5">ef7e7f67d</a>">ef7e7f67d</a><a href="https://github.com/taskforcesh/bullmq/commit/9acb18ce537a55a2fb60350ff216d15140aad68a"> into 1d83485e2">1d83485e2</a>

Run Details

621 of 845 branches covered (73.49%)

Branch coverage included in aggregate %.

4106 of 4532 relevant lines covered (90.6%)

74.87 hits per line

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

87.11

/src/test/test_pause.ts

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

describe('Pause', function() {
  let queue: Queue;
  let queueName: string;
  let queueEvents: QueueEvents;

  beforeEach(async function() {
    queueName = 'test-' + v4();
    queue = new Queue(queueName);
    queueEvents = new QueueEvents(queueName);
    await queueEvents.waitUntilReady();
  });

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

  // Skipped since some side effect makes this test fail
  it.skip('should not processed delayed jobs', async function() {
    this.timeout(5000);

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

    let processed = false;

    const worker = new Worker(queueName, async () => {
      processed = true;
    });
    await worker.waitUntilReady();

    await queue.pause();
    await queue.add('test', {}, { delay: 200 });
    const counts = await queue.getJobCounts('waiting', 'delayed');

    expect(counts).to.have.property('waiting', 0);
    expect(counts).to.have.property('delayed', 1);

    await delay(1000);
    if (processed) {
      throw new Error('should not process delayed jobs in paused queue.');
    }
    const counts2 = await queue.getJobCounts('waiting', 'paused', 'delayed');
    expect(counts2).to.have.property('waiting', 0);
    expect(counts2).to.have.property('paused', 1);
    expect(counts2).to.have.property('delayed', 0);

    await queueScheduler.close();
    await worker.close();
  });

  it('should pause a queue until resumed', async () => {
    let process;
    let isPaused = false;
    let counter = 2;
    const processPromise = new Promise(resolve => {
      process = async (job: Job) => {
        expect(isPaused).to.be.eql(false);
        expect(job.data.foo).to.be.equal('paused');
        counter--;
        if (counter === 0) {
          resolve();
        }
      };
    });

    const worker = new Worker(queueName, process);
    await worker.waitUntilReady();

    await queue.pause();
    isPaused = true;
    await queue.add('test', { foo: 'paused' });
    await queue.add('test', { foo: 'paused' });
    isPaused = false;
    await queue.resume();

    await processPromise;
    return worker.close();
  });

  it('should be able to pause a running queue and emit relevant events', async () => {
    let process;

    let isPaused = false,
      isResumed = true,
      first = true;

    const processPromise = new Promise((resolve, reject) => {
      process = async (job: Job) => {
        try {
          expect(isPaused).to.be.eql(false);
          expect(job.data.foo).to.be.equal('paused');

          if (first) {
            first = false;
            isPaused = true;
            return queue.pause();
          } else {
            expect(isResumed).to.be.eql(true);
            await queue.close();
            resolve();
          }
        } catch (err) {
          reject(err);
        }
      };
    });

    const worker = new Worker(queueName, process);

    queueEvents.on('paused', async () => {
      isPaused = false;
      await queue.resume();
    });

    queueEvents.on('resumed', () => {
      isResumed = true;
    });

    await queue.add('test', { foo: 'paused' });
    await queue.add('test', { foo: 'paused' });

    return processPromise;
  });

  it('should pause the queue locally', async () => {
    let worker: Worker;
    let counter = 2;
    let process;
    const processPromise = new Promise(resolve => {
      process = async (job: Job) => {
        expect(worker.isPaused()).to.be.eql(false);
        counter--;
        if (counter === 0) {
          await queue.close();
          resolve();
        }
      };
    });

    worker = new Worker(queueName, process);
    await worker.waitUntilReady();

    await worker.pause();

    // Add the worker after the queue is in paused mode since the normal behavior is to pause
    // it after the current lock expires. This way, we can ensure there isn't a lock already
    // to test that pausing behavior works.

    await queue.add('test', { foo: 'paused' });
    await queue.add('test', { foo: 'paused' });

    expect(counter).to.be.eql(2);
    expect(worker.isPaused()).to.be.eql(true);

    await worker.resume();

    return processPromise;
  });

  it('should wait until active jobs are finished before resolving pause', async () => {
    let process;

    const startProcessing = new Promise(resolve => {
      process = async () => {
        resolve();
        return delay(1000);
      };
    });

    const worker = new Worker(queueName, process);
    await worker.waitUntilReady();

    const jobs = [];
    for (let i = 0; i < 10; i++) {
      jobs.push(queue.add('test', i));
    }

    //
    // Add start processing so that we can test that pause waits for this job to be completed.
    //
    jobs.push(startProcessing);
    await Promise.all(jobs);
    await worker.pause();

    let active = await queue.getJobCountByTypes('active');
    expect(active).to.be.eql(0);
    expect(worker.isPaused()).to.be.eql(true);

    // One job from the 10 posted above will be processed, so we expect 9 jobs pending
    let paused = await queue.getJobCountByTypes('delayed', 'waiting');
    expect(paused).to.be.eql(9);

    await queue.add('test', {});

    active = await queue.getJobCountByTypes('active');
    expect(active).to.be.eql(0);

    paused = await queue.getJobCountByTypes('paused', 'waiting', 'delayed');
    expect(paused).to.be.eql(10);

    await worker.close();
  });

  it('should pause the queue locally when more than one worker is active', async () => {
    let process1, process2;

    const startProcessing1 = new Promise(resolve => {
      process1 = async () => {
        resolve();
        return delay(200);
      };
    });

    const startProcessing2 = new Promise(resolve => {
      process2 = async () => {
        resolve();
        return delay(200);
      };
    });

    const worker1 = new Worker(queueName, process1);
    await worker1.waitUntilReady();

    const worker2 = new Worker(queueName, process2);
    await worker2.waitUntilReady();

    await Promise.all([
      queue.add('test', 1),
      queue.add('test', 2),
      queue.add('test', 3),
      queue.add('test', 4),
    ]);

    await Promise.all([startProcessing1, startProcessing2]);
    await Promise.all([worker1.pause(), worker2.pause()]);

    const count = await queue.getJobCounts('active', 'waiting', 'completed');
    expect(count.active).to.be.eql(0);
    expect(count.waiting).to.be.eql(2);
    expect(count.completed).to.be.eql(2);

    return Promise.all([worker1.close(), worker2.close()]);
  });

  it('should wait for blocking job retrieval to complete before pausing locally', async () => {
    let process;

    const startProcessing = new Promise(resolve => {
      process = async () => {
        resolve();
        return delay(200);
      };
    });

    const worker = new Worker(queueName, process);
    await worker.waitUntilReady();

    await queue.add('test', 1);
    await startProcessing;
    await worker.pause();
    await queue.add('test', 2);

    const count = await queue.getJobCounts('active', 'waiting', 'completed');
    expect(count.active).to.be.eql(0);
    expect(count.waiting).to.be.eql(1);
    expect(count.completed).to.be.eql(1);

    return worker.close();
  });

  it('pauses fast when queue is drained', async function() {
    const worker = new Worker(queueName, async () => {});
    await worker.waitUntilReady();

    await queue.add('test', {});

    return new Promise((resolve, reject) => {
      queueEvents.on('drained', async () => {
        try {
          const start = new Date().getTime();
          await queue.pause();

          const finish = new Date().getTime();
          expect(finish - start).to.be.lt(1000);
        } catch (err) {
          reject(err);
        } finally {
          await worker.close();
        }
        resolve();
      });
    });
  });

  it('gets the right response from isPaused', async () => {
    await queue.pause();
    const isPausedQueuePaused = await queue.isPaused();
    expect(isPausedQueuePaused).to.be.true;

    await queue.resume();
    const isResumedQueuePaused = await queue.isPaused();
    expect(isResumedQueuePaused).to.be.false;
  });
});

1	import { Job, Queue } from '@src/classes';	1✔
2	import { QueueEvents, QueueScheduler } from '../classes';	1✔
3	import { Worker } from '@src/classes/worker';	1✔
4	import { expect } from 'chai';	1✔
5	import * as IORedis from 'ioredis';	1✔
6	import { beforeEach, describe, it } from 'mocha';	1✔
7	import { v4 } from 'uuid';	1✔
8	import { delay, removeAllQueueData } from '@src/utils';	1✔
9
10	describe('Pause', function() {	1✔
11	let queue: Queue;
12	let queueName: string;
13	let queueEvents: QueueEvents;
14
15	beforeEach(async function() {	1✔
16	queueName = 'test-' + v4();	8✔
17	queue = new Queue(queueName);	8✔
18	queueEvents = new QueueEvents(queueName);	8✔
19	await queueEvents.waitUntilReady();	8✔
20	});
21
22	afterEach(async function() {	1✔
23	await queue.close();	8✔
24	await queueEvents.close();	8✔
25	await removeAllQueueData(new IORedis(), queueName);	8✔
26	});
27
28	// Skipped since some side effect makes this test fail
29	it.skip('should not processed delayed jobs', async function() {	1✔
30	this.timeout(5000);	×
31
32	const queueScheduler = new QueueScheduler(queueName);	×
33	await queueScheduler.waitUntilReady();	×
34
35	let processed = false;	×
36
37	const worker = new Worker(queueName, async () => {	×
38	processed = true;	×
39	});
40	await worker.waitUntilReady();	×
41
42	await queue.pause();	×
43	await queue.add('test', {}, { delay: 200 });	×
44	const counts = await queue.getJobCounts('waiting', 'delayed');	×
45
46	expect(counts).to.have.property('waiting', 0);	×
47	expect(counts).to.have.property('delayed', 1);	×
48
49	await delay(1000);	×
50	if (processed) {	×
51	throw new Error('should not process delayed jobs in paused queue.');	×
52	}
53	const counts2 = await queue.getJobCounts('waiting', 'paused', 'delayed');	×
54	expect(counts2).to.have.property('waiting', 0);	×
55	expect(counts2).to.have.property('paused', 1);	×
56	expect(counts2).to.have.property('delayed', 0);	×
57
58	await queueScheduler.close();	×
59	await worker.close();	×
60	});
61
62	it('should pause a queue until resumed', async () => {	1✔
63	let process;
64	let isPaused = false;	1✔
65	let counter = 2;	1✔
66	const processPromise = new Promise(resolve => {	1✔
67	process = async (job: Job) => {	1✔
68	expect(isPaused).to.be.eql(false);	2✔
69	expect(job.data.foo).to.be.equal('paused');	2✔
70	counter--;	2✔
71	if (counter === 0) {	2✔
72	resolve();	1✔
73	}
74	};
75	});
76
77	const worker = new Worker(queueName, process);	1✔
78	await worker.waitUntilReady();	1✔
79
80	await queue.pause();	1✔
81	isPaused = true;	1✔
82	await queue.add('test', { foo: 'paused' });	1✔
83	await queue.add('test', { foo: 'paused' });	1✔
84	isPaused = false;	1✔
85	await queue.resume();	1✔
86
87	await processPromise;	1✔
88	return worker.close();	1✔
89	});
90
91	it('should be able to pause a running queue and emit relevant events', async () => {	1✔
92	let process;
93
94	let isPaused = false,	1✔
95	isResumed = true,	1✔
96	first = true;	1✔
97
98	const processPromise = new Promise((resolve, reject) => {	1✔
99	process = async (job: Job) => {	1✔
100	try {	2✔
101	expect(isPaused).to.be.eql(false);	2✔
102	expect(job.data.foo).to.be.equal('paused');	2✔
103
104	if (first) {	2✔
105	first = false;	1✔
106	isPaused = true;	1✔
107	return queue.pause();	1✔
108	} else {
109	expect(isResumed).to.be.eql(true);	1✔
110	await queue.close();	1✔
111	resolve();	1✔
112	}
113	} catch (err) {
114	reject(err);	×
115	}
116	};
117	});
118
119	const worker = new Worker(queueName, process);	1✔
120
121	queueEvents.on('paused', async () => {	1✔
122	isPaused = false;	1✔
123	await queue.resume();	1✔
124	});
125
126	queueEvents.on('resumed', () => {	1✔
127	isResumed = true;	1✔
128	});
129
130	await queue.add('test', { foo: 'paused' });	1✔
131	await queue.add('test', { foo: 'paused' });	1✔
132
133	return processPromise;	1✔
134	});
135
136	it('should pause the queue locally', async () => {	1✔
137	let worker: Worker;
138	let counter = 2;	1✔
139	let process;
140	const processPromise = new Promise(resolve => {	1✔
141	process = async (job: Job) => {	1✔
142	expect(worker.isPaused()).to.be.eql(false);	2✔
143	counter--;	2✔
144	if (counter === 0) {	2✔
145	await queue.close();	1✔
146	resolve();	1✔
147	}
148	};
149	});
150
151	worker = new Worker(queueName, process);	1✔
152	await worker.waitUntilReady();	1✔
153
154	await worker.pause();	1✔
155
156	// Add the worker after the queue is in paused mode since the normal behavior is to pause
157	// it after the current lock expires. This way, we can ensure there isn't a lock already
158	// to test that pausing behavior works.
159
160	await queue.add('test', { foo: 'paused' });	1✔
161	await queue.add('test', { foo: 'paused' });	1✔
162
163	expect(counter).to.be.eql(2);	1✔
164	expect(worker.isPaused()).to.be.eql(true);	1✔
165
166	await worker.resume();	1✔
167
168	return processPromise;	1✔
169	});
170
171	it('should wait until active jobs are finished before resolving pause', async () => {	1✔
172	let process;
173
174	const startProcessing = new Promise(resolve => {	1✔
175	process = async () => {	1✔
176	resolve();	1✔
177	return delay(1000);	1✔
178	};
179	});
180
181	const worker = new Worker(queueName, process);	1✔
182	await worker.waitUntilReady();	1✔
183
184	const jobs = [];	1✔
185	for (let i = 0; i < 10; i++) {	1✔
186	jobs.push(queue.add('test', i));	10✔
187	}
188
189	//
190	// Add start processing so that we can test that pause waits for this job to be completed.
191	//
192	jobs.push(startProcessing);	1✔
193	await Promise.all(jobs);	1✔
194	await worker.pause();	1✔
195
196	let active = await queue.getJobCountByTypes('active');	1✔
197	expect(active).to.be.eql(0);	1✔
198	expect(worker.isPaused()).to.be.eql(true);	1✔
199
200	// One job from the 10 posted above will be processed, so we expect 9 jobs pending
201	let paused = await queue.getJobCountByTypes('delayed', 'waiting');	1✔
202	expect(paused).to.be.eql(9);	1✔
203
204	await queue.add('test', {});	1✔
205
206	active = await queue.getJobCountByTypes('active');	1✔
207	expect(active).to.be.eql(0);	1✔
208
209	paused = await queue.getJobCountByTypes('paused', 'waiting', 'delayed');	1✔
210	expect(paused).to.be.eql(10);	1✔
211
212	await worker.close();	1✔
213	});
214
215	it('should pause the queue locally when more than one worker is active', async () => {	1✔
216	let process1, process2;
217
218	const startProcessing1 = new Promise(resolve => {	1✔
219	process1 = async () => {	1✔
220	resolve();	1✔
221	return delay(200);	1✔
222	};
223	});
224
225	const startProcessing2 = new Promise(resolve => {	1✔
226	process2 = async () => {	1✔
227	resolve();	1✔
228	return delay(200);	1✔
229	};
230	});
231
232	const worker1 = new Worker(queueName, process1);	1✔
233	await worker1.waitUntilReady();	1✔
234
235	const worker2 = new Worker(queueName, process2);	1✔
236	await worker2.waitUntilReady();	1✔
237
238	await Promise.all([	1✔
239	queue.add('test', 1),
240	queue.add('test', 2),
241	queue.add('test', 3),
242	queue.add('test', 4),
243	]);
244
245	await Promise.all([startProcessing1, startProcessing2]);	1✔
246	await Promise.all([worker1.pause(), worker2.pause()]);	1✔
247
248	const count = await queue.getJobCounts('active', 'waiting', 'completed');	1✔
249	expect(count.active).to.be.eql(0);	1✔
250	expect(count.waiting).to.be.eql(2);	1✔
251	expect(count.completed).to.be.eql(2);	1✔
252
253	return Promise.all([worker1.close(), worker2.close()]);	1✔
254	});
255
256	it('should wait for blocking job retrieval to complete before pausing locally', async () => {	1✔
257	let process;
258
259	const startProcessing = new Promise(resolve => {	1✔
260	process = async () => {	1✔
261	resolve();	1✔
262	return delay(200);	1✔
263	};
264	});
265
266	const worker = new Worker(queueName, process);	1✔
267	await worker.waitUntilReady();	1✔
268
269	await queue.add('test', 1);	1✔
270	await startProcessing;	1✔
271	await worker.pause();	1✔
272	await queue.add('test', 2);	1✔
273
274	const count = await queue.getJobCounts('active', 'waiting', 'completed');	1✔
275	expect(count.active).to.be.eql(0);	1✔
276	expect(count.waiting).to.be.eql(1);	1✔
277	expect(count.completed).to.be.eql(1);	1✔
278
279	return worker.close();	1✔
280	});
281
282	it('pauses fast when queue is drained', async function() {	1✔
283	const worker = new Worker(queueName, async () => {});	1✔
284	await worker.waitUntilReady();	1✔
285
286	await queue.add('test', {});	1✔
287
288	return new Promise((resolve, reject) => {	1✔
289	queueEvents.on('drained', async () => {	1✔
290	try {	1✔
291	const start = new Date().getTime();	1✔
292	await queue.pause();	1✔
293
294	const finish = new Date().getTime();	1✔
295	expect(finish - start).to.be.lt(1000);	1✔
296	} catch (err) {
297	reject(err);	×
298	} finally {
299	await worker.close();	1✔
300	}
301	resolve();	1✔
302	});
303	});
304	});
305
306	it('gets the right response from isPaused', async () => {	1✔
307	await queue.pause();	1✔
308	const isPausedQueuePaused = await queue.isPaused();	1✔
309	expect(isPausedQueuePaused).to.be.true;	1✔
310
311	await queue.resume();	1✔
312	const isResumedQueuePaused = await queue.isPaused();	1✔
313	expect(isResumedQueuePaused).to.be.false;	1✔
314	});
315	});

taskforcesh / bullmq / #269

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