8811069601

Committed 24 Apr 2024 04:33AM UTC coverage: 88.979% (-0.004%) from 88.983%

Build # 8811069601

Build Type

Pull #2496

github

Committed by

web-flow

Commit Message

Merge 2a200e71b into 0140959ca

Pull Request Pull Request #2496: feat(queue): add global concurrency

Run Details

1110 of 1344 branches covered (82.59%)

Branch coverage included in aggregate %.

9 of 9 new or added lines in 3 files covered. (100.0%)

7 existing lines in 2 files now uncovered.

2168 of 2340 relevant lines covered (92.65%)

2966.58 hits per line

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

93.38

/src/classes/queue.ts

import { v4 } from 'uuid';
import {
  BaseJobOptions,
  BulkJobOptions,
  IoredisListener,
  QueueOptions,
  RepeatableJob,
  RepeatOptions,
} from '../interfaces';
import { FinishedStatus, JobsOptions, MinimalQueue } from '../types';
import { Job } from './job';
import { QueueGetters } from './queue-getters';
import { Repeat } from './repeat';
import { RedisConnection } from './redis-connection';

export interface ObliterateOpts {
  /**
   * Use force = true to force obliteration even with active jobs in the queue
   * @defaultValue false
   */
  force?: boolean;
  /**
   * Use count with the maximum number of deleted keys per iteration
   * @defaultValue 1000
   */
  count?: number;
}

export interface QueueListener<DataType, ResultType, NameType extends string>
  extends IoredisListener {
  /**
   * Listen to 'cleaned' event.
   *
   * This event is triggered when the queue calls clean method.
   */
  cleaned: (jobs: string[], type: string) => void;

  /**
   * Listen to 'error' event.
   *
   * This event is triggered when an error is thrown.
   */
  error: (err: Error) => void;

  /**
   * Listen to 'paused' event.
   *
   * This event is triggered when the queue is paused.
   */
  paused: () => void;

  /**
   * Listen to 'progress' event.
   *
   * This event is triggered when the job updates its progress.
   */
  progress: (
    job: Job<DataType, ResultType, NameType>,
    progress: number | object,
  ) => void;

  /**
   * Listen to 'removed' event.
   *
   * This event is triggered when a job is removed.
   */
  removed: (job: Job<DataType, ResultType, NameType>) => void;

  /**
   * Listen to 'resumed' event.
   *
   * This event is triggered when the queue is resumed.
   */
  resumed: () => void;

  /**
   * Listen to 'waiting' event.
   *
   * This event is triggered when the queue creates a new job.
   */
  waiting: (job: Job<DataType, ResultType, NameType>) => void;
}

/**
 * Queue
 *
 * This class provides methods to add jobs to a queue and some other high-level
 * administration such as pausing or deleting queues.
 *
 */
export class Queue<
  DataType = any,
  ResultType = any,
  NameType extends string = string,
> extends QueueGetters<DataType, ResultType, NameType> {
  token = v4();
  jobsOpts: BaseJobOptions;
  opts: QueueOptions;
  private _repeat?: Repeat;

  constructor(
    name: string,
    opts?: QueueOptions,
    Connection?: typeof RedisConnection,
  ) {
    super(
      name,
      {
        blockingConnection: false,
        ...opts,
      },
      Connection,
    );

    this.jobsOpts = opts?.defaultJobOptions ?? {};

    this.waitUntilReady()
      .then(client => {
        if (!this.closing) {
          client.hmset(this.keys.meta, this.metaValues);
        }
      })
      .catch(err => {
        // We ignore this error to avoid warnings. The error can still
        // be received by listening to event 'error'
      });
  }

  emit<U extends keyof QueueListener<DataType, ResultType, NameType>>(
    event: U,
    ...args: Parameters<QueueListener<DataType, ResultType, NameType>[U]>
  ): boolean {
    return super.emit(event, ...args);
  }

  off<U extends keyof QueueListener<DataType, ResultType, NameType>>(
    eventName: U,
    listener: QueueListener<DataType, ResultType, NameType>[U],
  ): this {
    super.off(eventName, listener);
    return this;
  }

  on<U extends keyof QueueListener<DataType, ResultType, NameType>>(
    event: U,
    listener: QueueListener<DataType, ResultType, NameType>[U],
  ): this {
    super.on(event, listener);
    return this;
  }

  once<U extends keyof QueueListener<DataType, ResultType, NameType>>(
    event: U,
    listener: QueueListener<DataType, ResultType, NameType>[U],
  ): this {
    super.once(event, listener);
    return this;
  }

  /**
   * Returns this instance current default job options.
   */
  get defaultJobOptions(): JobsOptions {
    return { ...this.jobsOpts };
  }

  get metaValues(): Record<string, string | number> {
    return {
      'opts.maxLenEvents': this.opts?.streams?.events?.maxLen ?? 10000,
    };
  }

  get repeat(): Promise<Repeat> {
    return new Promise<Repeat>(async resolve => {
      if (!this._repeat) {
        this._repeat = new Repeat(this.name, {
          ...this.opts,
          connection: await this.client,
        });
        this._repeat.on('error', e => this.emit.bind(this, e));
      }
      resolve(this._repeat);
    });
  }

  /**
   * Enable and set global concurrency value.
   * @param concurrency - Maximum number of simultaneous jobs that the workers can handle.
   * For instance, setting this value to 1 ensures that no more than one job
   * is processed at any given time. If this limit is not defined, there will be no
   * restriction on the number of concurrent jobs.
   */
  async setGlobalConcurrency(concurrency: number) {
    const client = await this.client;
    return client.hset(this.keys.meta, 'concurrency', concurrency);
  }

  /**
   * Adds a new job to the queue.
   *
   * @param name - Name of the job to be added to the queue,.
   * @param data - Arbitrary data to append to the job.
   * @param opts - Job options that affects how the job is going to be processed.
   */
  async add(
    name: NameType,
    data: DataType,
    opts?: JobsOptions,
  ): Promise<Job<DataType, ResultType, NameType>> {
    if (opts && opts.repeat) {
      return (await this.repeat).addNextRepeatableJob<
        DataType,
        ResultType,
        NameType
      >(name, data, { ...this.jobsOpts, ...opts }, true);
    } else {
      const jobId = opts?.jobId;

      if (jobId == '0' || jobId?.startsWith('0:')) {
        throw new Error("JobId cannot be '0' or start with 0:");
      }

      const job = await this.Job.create<DataType, ResultType, NameType>(
        this as MinimalQueue,
        name,
        data,
        {
          ...this.jobsOpts,
          ...opts,
          jobId,
        },
      );
      this.emit('waiting', job);
      return job;
    }
  }

  /**
   * Adds an array of jobs to the queue. This method may be faster than adding
   * one job at a time in a sequence.
   *
   * @param jobs - The array of jobs to add to the queue. Each job is defined by 3
   * properties, 'name', 'data' and 'opts'. They follow the same signature as 'Queue.add'.
   */
  addBulk(
    jobs: { name: NameType; data: DataType; opts?: BulkJobOptions }[],
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.Job.createBulk<DataType, ResultType, NameType>(
      this as MinimalQueue,
      jobs.map(job => ({
        name: job.name,
        data: job.data,
        opts: {
          ...this.jobsOpts,
          ...job.opts,
          jobId: job.opts?.jobId,
        },
      })),
    );
  }

  /**
   * Pauses the processing of this queue globally.
   *
   * We use an atomic RENAME operation on the wait queue. Since
   * we have blocking calls with BRPOPLPUSH on the wait queue, as long as the queue
   * is renamed to 'paused', no new jobs will be processed (the current ones
   * will run until finalized).
   *
   * Adding jobs requires a LUA script to check first if the paused list exist
   * and in that case it will add it there instead of the wait list.
   */
  async pause(): Promise<void> {
    await this.scripts.pause(true);
    this.emit('paused');
  }

  /**
   * Close the queue instance.
   *
   */
  async close(): Promise<void> {
    if (!this.closing) {
      if (this._repeat) {
        await this._repeat.close();
      }
    }
    return super.close();
  }
  /**
   * Resumes the processing of this queue globally.
   *
   * The method reverses the pause operation by resuming the processing of the
   * queue.
   */
  async resume(): Promise<void> {
    await this.scripts.pause(false);
    this.emit('resumed');
  }

  /**
   * Returns true if the queue is currently paused.
   */
  async isPaused(): Promise<boolean> {
    const client = await this.client;
    const pausedKeyExists = await client.hexists(this.keys.meta, 'paused');
    return pausedKeyExists === 1;
  }

  /**
   * Returns true if the queue is currently maxed.
   */
  async isMaxed(): Promise<boolean> {
    const client = await this.client;
    const maxed = await client.hexists(this.keys.meta, 'maxed');
    return maxed === 1;
  }

  /**
   * Get all repeatable meta jobs.
   *
   * @param start - Offset of first job to return.
   * @param end - Offset of last job to return.
   * @param asc - Determine the order in which jobs are returned based on their
   * next execution time.
   */
  async getRepeatableJobs(
    start?: number,
    end?: number,
    asc?: boolean,
  ): Promise<RepeatableJob[]> {
    return (await this.repeat).getRepeatableJobs(start, end, asc);
  }

  /**
   * Removes a repeatable job.
   *
   * Note: you need to use the exact same repeatOpts when deleting a repeatable job
   * than when adding it.
   *
   * @see removeRepeatableByKey
   *
   * @param name - job name
   * @param repeatOpts -
   * @param jobId -
   * @returns
   */
  async removeRepeatable(
    name: NameType,
    repeatOpts: RepeatOptions,
    jobId?: string,
  ): Promise<boolean> {
    const repeat = await this.repeat;
    const removed = await repeat.removeRepeatable(name, repeatOpts, jobId);

    return !removed;
  }

  /**
   * Removes a repeatable job by its key. Note that the key is the one used
   * to store the repeatable job metadata and not one of the job iterations
   * themselves. You can use "getRepeatableJobs" in order to get the keys.
   *
   * @see getRepeatableJobs
   *
   * @param repeatJobKey - to the repeatable job.
   * @returns
   */
  async removeRepeatableByKey(key: string): Promise<boolean> {
    const repeat = await this.repeat;
    const removed = await repeat.removeRepeatableByKey(key);

    return !removed;
  }

  /**
   * Removes the given job from the queue as well as all its
   * dependencies.
   *
   * @param jobId - The id of the job to remove
   * @param opts - Options to remove a job
   * @returns 1 if it managed to remove the job or 0 if the job or
   * any of its dependencies were locked.
   */
  remove(jobId: string, { removeChildren = true } = {}): Promise<number> {
    return this.scripts.remove(jobId, removeChildren);
  }

  /**
   * Updates the given job's progress.
   *
   * @param jobId - The id of the job to update
   * @param progress - number or object to be saved as progress.
   */
  async updateJobProgress(
    jobId: string,
    progress: number | object,
  ): Promise<void> {
    return this.scripts.updateProgress(jobId, progress);
  }

  /**
   * Logs one row of job's log data.
   *
   * @param jobId - The job id to log against.
   * @param logRow - string with log data to be logged.
   * @param keepLogs - max number of log entries to keep (0 for unlimited).
   *
   * @returns The total number of log entries for this job so far.
   */
  async addJobLog(
    jobId: string,
    logRow: string,
    keepLogs?: number,
  ): Promise<number> {
    return Job.addJobLog(this, jobId, logRow, keepLogs);
  }

  /**
   * Drains the queue, i.e., removes all jobs that are waiting
   * or delayed, but not active, completed or failed.
   *
   * @param delayed - Pass true if it should also clean the
   * delayed jobs.
   */
  drain(delayed = false): Promise<void> {
    return this.scripts.drain(delayed);
  }

  /**
   * Cleans jobs from a queue. Similar to drain but keeps jobs within a certain
   * grace period.
   *
   * @param grace - The grace period
   * @param limit - Max number of jobs to clean
   * @param type - The type of job to clean
   * Possible values are completed, wait, active, paused, delayed, failed. Defaults to completed.
   * @returns Id jobs from the deleted records
   */
  async clean(
    grace: number,
    limit: number,
    type:
      | 'completed'
      | 'wait'
      | 'active'
      | 'paused'
      | 'prioritized'
      | 'delayed'
      | 'failed' = 'completed',
  ): Promise<string[]> {
    const maxCount = limit || Infinity;
    const maxCountPerCall = Math.min(10000, maxCount);
    const timestamp = Date.now() - grace;
    let deletedCount = 0;
    const deletedJobsIds: string[] = [];

    while (deletedCount < maxCount) {
      const jobsIds = await this.scripts.cleanJobsInSet(
        type,
        timestamp,
        maxCountPerCall,
      );

      this.emit('cleaned', jobsIds, type);
      deletedCount += jobsIds.length;
      deletedJobsIds.push(...jobsIds);

      if (jobsIds.length < maxCountPerCall) {
        break;
      }
    }
    return deletedJobsIds;
  }

  /**
   * Completely destroys the queue and all of its contents irreversibly.
   * This method will the *pause* the queue and requires that there are no
   * active jobs. It is possible to bypass this requirement, i.e. not
   * having active jobs using the "force" option.
   *
   * Note: This operation requires to iterate on all the jobs stored in the queue
   * and can be slow for very large queues.
   *
   * @param opts - Obliterate options.
   */
  async obliterate(opts?: ObliterateOpts): Promise<void> {
    await this.pause();

    let cursor = 0;
    do {
      cursor = await this.scripts.obliterate({
        force: false,
        count: 1000,
        ...opts,
      });
    } while (cursor);
  }

  /**
   * Retry all the failed or completed jobs.
   *
   * @param opts: { count: number; state: FinishedStatus; timestamp: number}
   *   - count  number to limit how many jobs will be moved to wait status per iteration,
   *   - state  failed by default or completed.
   *   - timestamp from which timestamp to start moving jobs to wait status, default Date.now().
   *
   * @returns
   */
  async retryJobs(
    opts: { count?: number; state?: FinishedStatus; timestamp?: number } = {},
  ): Promise<void> {
    let cursor = 0;
    do {
      cursor = await this.scripts.retryJobs(
        opts.state,
        opts.count,
        opts.timestamp,
      );
    } while (cursor);
  }

  /**
   * Promote all the delayed jobs.
   *
   * @param opts: { count: number }
   *   - count  number to limit how many jobs will be moved to wait status per iteration
   *
   * @returns
   */
  async promoteJobs(opts: { count?: number } = {}): Promise<void> {
    let cursor = 0;
    do {
      cursor = await this.scripts.promoteJobs(opts.count);
    } while (cursor);
  }

  /**
   * Trim the event stream to an approximately maxLength.
   *
   * @param maxLength -
   */
  async trimEvents(maxLength: number): Promise<number> {
    const client = await this.client;
    return client.xtrim(this.keys.events, 'MAXLEN', '~', maxLength);
  }

  /**
   * Delete old priority helper key.
   */
  async removeDeprecatedPriorityKey(): Promise<number> {
    const client = await this.client;
    return client.del(this.toKey('priority'));
  }
}

1	import { v4 } from 'uuid';	1✔
2	import {
3	BaseJobOptions,
4	BulkJobOptions,
5	IoredisListener,
6	QueueOptions,
7	RepeatableJob,
8	RepeatOptions,
9	} from '../interfaces';
10	import { FinishedStatus, JobsOptions, MinimalQueue } from '../types';
11	import { Job } from './job';	1✔
12	import { QueueGetters } from './queue-getters';	1✔
13	import { Repeat } from './repeat';	1✔
14	import { RedisConnection } from './redis-connection';
15
16	export interface ObliterateOpts {
17	/**
18	* Use force = true to force obliteration even with active jobs in the queue
19	* @defaultValue false
20	*/
21	force?: boolean;
22	/**
23	* Use count with the maximum number of deleted keys per iteration
24	* @defaultValue 1000
25	*/
26	count?: number;
27	}
28
29	export interface QueueListener<DataType, ResultType, NameType extends string>
30	extends IoredisListener {
31	/**
32	* Listen to 'cleaned' event.
33	*
34	* This event is triggered when the queue calls clean method.
35	*/
36	cleaned: (jobs: string[], type: string) => void;
37
38	/**
39	* Listen to 'error' event.
40	*
41	* This event is triggered when an error is thrown.
42	*/
43	error: (err: Error) => void;
44
45	/**
46	* Listen to 'paused' event.
47	*
48	* This event is triggered when the queue is paused.
49	*/
50	paused: () => void;
51
52	/**
53	* Listen to 'progress' event.
54	*
55	* This event is triggered when the job updates its progress.
56	*/
57	progress: (
58	job: Job<DataType, ResultType, NameType>,
59	progress: number \| object,
60	) => void;
61
62	/**
63	* Listen to 'removed' event.
64	*
65	* This event is triggered when a job is removed.
66	*/
67	removed: (job: Job<DataType, ResultType, NameType>) => void;
68
69	/**
70	* Listen to 'resumed' event.
71	*
72	* This event is triggered when the queue is resumed.
73	*/
74	resumed: () => void;
75
76	/**
77	* Listen to 'waiting' event.
78	*
79	* This event is triggered when the queue creates a new job.
80	*/
81	waiting: (job: Job<DataType, ResultType, NameType>) => void;
82	}
83
84	/**
85	* Queue
86	*
87	* This class provides methods to add jobs to a queue and some other high-level
88	* administration such as pausing or deleting queues.
89	*
90	*/
91	export class Queue<	1✔
92	DataType = any,
93	ResultType = any,
94	NameType extends string = string,
95	> extends QueueGetters<DataType, ResultType, NameType> {
96	token = v4();	551✔
97	jobsOpts: BaseJobOptions;
98	opts: QueueOptions;
99	private _repeat?: Repeat;
100
101	constructor(
102	name: string,
103	opts?: QueueOptions,
104	Connection?: typeof RedisConnection,
105	) {
106	super(	552✔
107	name,
108	{
109	blockingConnection: false,
110	...opts,
111	},
112	Connection,
113	);
114
115	this.jobsOpts = opts?.defaultJobOptions ?? {};	551!
116
117	this.waitUntilReady()	551✔
118	.then(client => {
119	if (!this.closing) {	546✔
120	client.hmset(this.keys.meta, this.metaValues);	544✔
121	}
122	})
123	.catch(err => {
124	// We ignore this error to avoid warnings. The error can still
125	// be received by listening to event 'error'
126	});
127	}
128
129	emit<U extends keyof QueueListener<DataType, ResultType, NameType>>(
130	event: U,
131	...args: Parameters<QueueListener<DataType, ResultType, NameType>[U]>
132	): boolean {
133	return super.emit(event, ...args);	11,652✔
134	}
135
136	off<U extends keyof QueueListener<DataType, ResultType, NameType>>(
137	eventName: U,
138	listener: QueueListener<DataType, ResultType, NameType>[U],
139	): this {
140	super.off(eventName, listener);	×
141	return this;	×
142	}
143
144	on<U extends keyof QueueListener<DataType, ResultType, NameType>>(
145	event: U,
146	listener: QueueListener<DataType, ResultType, NameType>[U],
147	): this {
148	super.on(event, listener);	33✔
149	return this;	33✔
150	}
151
152	once<U extends keyof QueueListener<DataType, ResultType, NameType>>(
153	event: U,
154	listener: QueueListener<DataType, ResultType, NameType>[U],
155	): this {
156	super.once(event, listener);	×
157	return this;	×
158	}
159
160	/**
161	* Returns this instance current default job options.
162	*/
163	get defaultJobOptions(): JobsOptions {
164	return { ...this.jobsOpts };	×
165	}
166
167	get metaValues(): Record<string, string \| number> {
168	return {	544✔
169	'opts.maxLenEvents': this.opts?.streams?.events?.maxLen ?? 10000,	6,528!
170	};
171	}
172
173	get repeat(): Promise<Repeat> {
174	return new Promise<Repeat>(async resolve => {	51✔
175	if (!this._repeat) {	51✔
176	this._repeat = new Repeat(this.name, {	36✔
177	...this.opts,
178	connection: await this.client,
179	});
180	this._repeat.on('error', e => this.emit.bind(this, e));	36✔
181	}
182	resolve(this._repeat);	51✔
183	});
184	}
185
186	/**
187	* Enable and set global concurrency value.
188	* @param concurrency - Maximum number of simultaneous jobs that the workers can handle.
189	* For instance, setting this value to 1 ensures that no more than one job
190	* is processed at any given time. If this limit is not defined, there will be no
191	* restriction on the number of concurrent jobs.
192	*/
193	async setGlobalConcurrency(concurrency: number) {
194	const client = await this.client;	9✔
195	return client.hset(this.keys.meta, 'concurrency', concurrency);	9✔
196	}
197
198	/**
199	* Adds a new job to the queue.
200	*
201	* @param name - Name of the job to be added to the queue,.
202	* @param data - Arbitrary data to append to the job.
203	* @param opts - Job options that affects how the job is going to be processed.
204	*/
205	async add(
206	name: NameType,
207	data: DataType,
208	opts?: JobsOptions,
209	): Promise<Job<DataType, ResultType, NameType>> {
210	if (opts && opts.repeat) {	3,591✔
211	return (await this.repeat).addNextRepeatableJob<	38✔
212	DataType,
213	ResultType,
214	NameType
215	>(name, data, { ...this.jobsOpts, ...opts }, true);
216	} else {
217	const jobId = opts?.jobId;	3,553✔
218
219	if (jobId == '0' \|\| jobId?.startsWith('0:')) {	3,553!
UNCOV 220	throw new Error("JobId cannot be '0' or start with 0:");	×
221	}
222
223	const job = await this.Job.create<DataType, ResultType, NameType>(	3,553✔
224	this as MinimalQueue,
225	name,
226	data,
227	{
228	...this.jobsOpts,
229	...opts,
230	jobId,
231	},
232	);
233	this.emit('waiting', job);	3,551✔
234	return job;	3,551✔
235	}
236	}
237
238	/**
239	* Adds an array of jobs to the queue. This method may be faster than adding
240	* one job at a time in a sequence.
241	*
242	* @param jobs - The array of jobs to add to the queue. Each job is defined by 3
243	* properties, 'name', 'data' and 'opts'. They follow the same signature as 'Queue.add'.
244	*/
245	addBulk(
246	jobs: { name: NameType; data: DataType; opts?: BulkJobOptions }[],
247	): Promise<Job<DataType, ResultType, NameType>[]> {
248	return this.Job.createBulk<DataType, ResultType, NameType>(	45✔
249	this as MinimalQueue,
250	jobs.map(job => ({	9,142✔
251	name: job.name,
252	data: job.data,
253	opts: {
254	...this.jobsOpts,
255	...job.opts,
256	jobId: job.opts?.jobId,	27,426✔
257	},
258	})),
259	);
260	}
261
262	/**
263	* Pauses the processing of this queue globally.
264	*
265	* We use an atomic RENAME operation on the wait queue. Since
266	* we have blocking calls with BRPOPLPUSH on the wait queue, as long as the queue
267	* is renamed to 'paused', no new jobs will be processed (the current ones
268	* will run until finalized).
269	*
270	* Adding jobs requires a LUA script to check first if the paused list exist
271	* and in that case it will add it there instead of the wait list.
272	*/
273	async pause(): Promise<void> {
274	await this.scripts.pause(true);	35✔
275	this.emit('paused');	35✔
276	}
277
278	/**
279	* Close the queue instance.
280	*
281	*/
282	async close(): Promise<void> {
283	if (!this.closing) {	547✔
284	if (this._repeat) {	545✔
285	await this._repeat.close();	30✔
286	}
287	}
288	return super.close();	547✔
289	}
290	/**
291	* Resumes the processing of this queue globally.
292	*
293	* The method reverses the pause operation by resuming the processing of the
294	* queue.
295	*/
296	async resume(): Promise<void> {
297	await this.scripts.pause(false);	13✔
298	this.emit('resumed');	13✔
299	}
300
301	/**
302	* Returns true if the queue is currently paused.
303	*/
304	async isPaused(): Promise<boolean> {
305	const client = await this.client;	2✔
306	const pausedKeyExists = await client.hexists(this.keys.meta, 'paused');	2✔
307	return pausedKeyExists === 1;	2✔
308	}
309
310	/**
311	* Returns true if the queue is currently maxed.
312	*/
313	async isMaxed(): Promise<boolean> {
314	const client = await this.client;	2✔
315	const maxed = await client.hexists(this.keys.meta, 'maxed');	2✔
316	return maxed === 1;	2✔
317	}
318
319	/**
320	* Get all repeatable meta jobs.
321	*
322	* @param start - Offset of first job to return.
323	* @param end - Offset of last job to return.
324	* @param asc - Determine the order in which jobs are returned based on their
325	* next execution time.
326	*/
327	async getRepeatableJobs(
328	start?: number,
329	end?: number,
330	asc?: boolean,
331	): Promise<RepeatableJob[]> {
332	return (await this.repeat).getRepeatableJobs(start, end, asc);	4✔
333	}
334
335	/**
336	* Removes a repeatable job.
337	*
338	* Note: you need to use the exact same repeatOpts when deleting a repeatable job
339	* than when adding it.
340	*
341	* @see removeRepeatableByKey
342	*
343	* @param name - job name
344	* @param repeatOpts -
345	* @param jobId -
346	* @returns
347	*/
348	async removeRepeatable(
349	name: NameType,
350	repeatOpts: RepeatOptions,
351	jobId?: string,
352	): Promise<boolean> {
353	const repeat = await this.repeat;	6✔
354	const removed = await repeat.removeRepeatable(name, repeatOpts, jobId);	6✔
355
356	return !removed;	6✔
357	}
358
359	/**
360	* Removes a repeatable job by its key. Note that the key is the one used
361	* to store the repeatable job metadata and not one of the job iterations
362	* themselves. You can use "getRepeatableJobs" in order to get the keys.
363	*
364	* @see getRepeatableJobs
365	*
366	* @param repeatJobKey - to the repeatable job.
367	* @returns
368	*/
369	async removeRepeatableByKey(key: string): Promise<boolean> {
370	const repeat = await this.repeat;	3✔
371	const removed = await repeat.removeRepeatableByKey(key);	3✔
372
373	return !removed;	3✔
374	}
375
376	/**
377	* Removes the given job from the queue as well as all its
378	* dependencies.
379	*
380	* @param jobId - The id of the job to remove
381	* @param opts - Options to remove a job
382	* @returns 1 if it managed to remove the job or 0 if the job or
383	* any of its dependencies were locked.
384	*/
385	remove(jobId: string, { removeChildren = true } = {}): Promise<number> {	600✔
386	return this.scripts.remove(jobId, removeChildren);	300✔
387	}
388
389	/**
390	* Updates the given job's progress.
391	*
392	* @param jobId - The id of the job to update
393	* @param progress - number or object to be saved as progress.
394	*/
395	async updateJobProgress(
396	jobId: string,
397	progress: number \| object,
398	): Promise<void> {
399	return this.scripts.updateProgress(jobId, progress);	2✔
400	}
401
402	/**
403	* Logs one row of job's log data.
404	*
405	* @param jobId - The job id to log against.
406	* @param logRow - string with log data to be logged.
407	* @param keepLogs - max number of log entries to keep (0 for unlimited).
408	*
409	* @returns The total number of log entries for this job so far.
410	*/
411	async addJobLog(
412	jobId: string,
413	logRow: string,
414	keepLogs?: number,
415	): Promise<number> {
416	return Job.addJobLog(this, jobId, logRow, keepLogs);	2✔
417	}
418
419	/**
420	* Drains the queue, i.e., removes all jobs that are waiting
421	* or delayed, but not active, completed or failed.
422	*
423	* @param delayed - Pass true if it should also clean the
424	* delayed jobs.
425	*/
426	drain(delayed = false): Promise<void> {	7✔
427	return this.scripts.drain(delayed);	9✔
428	}
429
430	/**
431	* Cleans jobs from a queue. Similar to drain but keeps jobs within a certain
432	* grace period.
433	*
434	* @param grace - The grace period
435	* @param limit - Max number of jobs to clean
436	* @param type - The type of job to clean
437	* Possible values are completed, wait, active, paused, delayed, failed. Defaults to completed.
438	* @returns Id jobs from the deleted records
439	*/
440	async clean(
441	grace: number,
442	limit: number,
443	type:	4✔
444	\| 'completed'
445	\| 'wait'
446	\| 'active'
447	\| 'paused'
448	\| 'prioritized'
449	\| 'delayed'
450	\| 'failed' = 'completed',
451	): Promise<string[]> {
452	const maxCount = limit \|\| Infinity;	24✔
453	const maxCountPerCall = Math.min(10000, maxCount);	24✔
454	const timestamp = Date.now() - grace;	24✔
455	let deletedCount = 0;	24✔
456	const deletedJobsIds: string[] = [];	24✔
457
458	while (deletedCount < maxCount) {	24✔
459	const jobsIds = await this.scripts.cleanJobsInSet(	24✔
460	type,
461	timestamp,
462	maxCountPerCall,
463	);
464
465	this.emit('cleaned', jobsIds, type);	24✔
466	deletedCount += jobsIds.length;	24✔
467	deletedJobsIds.push(...jobsIds);	24✔
468
469	if (jobsIds.length < maxCountPerCall) {	24✔
470	break;	23✔
471	}
472	}
473	return deletedJobsIds;	24✔
474	}
475
476	/**
477	* Completely destroys the queue and all of its contents irreversibly.
478	* This method will the pause the queue and requires that there are no
479	* active jobs. It is possible to bypass this requirement, i.e. not
480	* having active jobs using the "force" option.
481	*
482	* Note: This operation requires to iterate on all the jobs stored in the queue
483	* and can be slow for very large queues.
484	*
485	* @param opts - Obliterate options.
486	*/
487	async obliterate(opts?: ObliterateOpts): Promise<void> {
488	await this.pause();	12✔
489
490	let cursor = 0;	12✔
491	do {	12✔
492	cursor = await this.scripts.obliterate({	14✔
493	force: false,
494	count: 1000,
495	...opts,
496	});
497	} while (cursor);
498	}
499
500	/**
501	* Retry all the failed or completed jobs.
502	*
503	* @param opts: { count: number; state: FinishedStatus; timestamp: number}
504	* - count number to limit how many jobs will be moved to wait status per iteration,
505	* - state failed by default or completed.
506	* - timestamp from which timestamp to start moving jobs to wait status, default Date.now().
507	*
508	* @returns
509	*/
510	async retryJobs(
511	opts: { count?: number; state?: FinishedStatus; timestamp?: number } = {},	×
512	): Promise<void> {
513	let cursor = 0;	4✔
514	do {	4✔
515	cursor = await this.scripts.retryJobs(	18✔
516	opts.state,
517	opts.count,
518	opts.timestamp,
519	);
520	} while (cursor);
521	}
522
523	/**
524	* Promote all the delayed jobs.
525	*
526	* @param opts: { count: number }
527	* - count number to limit how many jobs will be moved to wait status per iteration
528	*
529	* @returns
530	*/
531	async promoteJobs(opts: { count?: number } = {}): Promise<void> {	1✔
532	let cursor = 0;	1✔
533	do {	1✔
534	cursor = await this.scripts.promoteJobs(opts.count);	1✔
535	} while (cursor);
536	}
537
538	/**
539	* Trim the event stream to an approximately maxLength.
540	*
541	* @param maxLength -
542	*/
543	async trimEvents(maxLength: number): Promise<number> {
544	const client = await this.client;	1✔
545	return client.xtrim(this.keys.events, 'MAXLEN', '~', maxLength);	1✔
546	}
547
548	/**
549	* Delete old priority helper key.
550	*/
551	async removeDeprecatedPriorityKey(): Promise<number> {
552	const client = await this.client;	1✔
553	return client.del(this.toKey('priority'));	1✔
554	}
555	}

taskforcesh / bullmq / 8811069601

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