11219305157

Committed 07 Oct 2024 03:54PM UTC coverage: 88.751% (-0.1%) from 88.856%

Build # 11219305157

Build Type

push

github

Committed by

web-flow

Commit Message

feat(job): add deduplication logic (#2796)

Run Details

1267 of 1523 branches covered (83.19%)

Branch coverage included in aggregate %.

3 of 7 new or added lines in 3 files covered. (42.86%)

3 existing lines in 2 files now uncovered.

2370 of 2575 relevant lines covered (92.04%)

4189.57 hits per line

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

90.05

/src/classes/queue-getters.ts

/*eslint-env node */
'use strict';

import { QueueBase } from './queue-base';
import { Job } from './job';
import { clientCommandMessageReg, QUEUE_EVENT_SUFFIX } from '../utils';
import { JobState, JobType } from '../types';
import { JobJsonRaw, Metrics } from '../interfaces';

/**
 *
 * @class QueueGetters
 * @extends QueueBase
 *
 * @description Provides different getters for different aspects of a queue.
 */
export class QueueGetters<
  DataType,
  ResultType,
  NameType extends string,
> extends QueueBase {
  getJob(
    jobId: string,
  ): Promise<Job<DataType, ResultType, NameType> | undefined> {
    return this.Job.fromId(this, jobId) as Promise<
      Job<DataType, ResultType, NameType>
    >;
  }

  private commandByType(
    types: JobType[],
    count: boolean,
    callback: (key: string, dataType: string) => void,
  ) {
    return types.map((type: string) => {
      type = type === 'waiting' ? 'wait' : type; // alias

      const key = this.toKey(type);

      switch (type) {
        case 'completed':
        case 'failed':
        case 'delayed':
        case 'prioritized':
        case 'repeat':
        case 'waiting-children':
          return callback(key, count ? 'zcard' : 'zrange');
        case 'active':
        case 'wait':
        case 'paused':
          return callback(key, count ? 'llen' : 'lrange');
      }
    });
  }

  /**
   * Helper to easily extend Job class calls.
   */
  protected get Job(): typeof Job {
    return Job;
  }

  private sanitizeJobTypes(types: JobType[] | JobType | undefined): JobType[] {
    const currentTypes = typeof types === 'string' ? [types] : types;

    if (Array.isArray(currentTypes) && currentTypes.length > 0) {
      const sanitizedTypes = [...currentTypes];

      if (sanitizedTypes.indexOf('waiting') !== -1) {
        sanitizedTypes.push('paused');
      }

      return [...new Set(sanitizedTypes)];
    }

    return [
      'active',
      'completed',
      'delayed',
      'failed',
      'paused',
      'prioritized',
      'waiting',
      'waiting-children',
    ];
  }

  /**
    Returns the number of jobs waiting to be processed. This includes jobs that are
    "waiting" or "delayed" or "prioritized" or "waiting-children".
  */
  async count(): Promise<number> {
    const count = await this.getJobCountByTypes(
      'waiting',
      'paused',
      'delayed',
      'prioritized',
      'waiting-children',
    );

    return count;
  }

  /**
   * Returns the time to live for a rate limited key in milliseconds.
   * @param maxJobs - max jobs to be considered in rate limit state. If not passed
   * it will return the remaining ttl without considering if max jobs is excedeed.
   * @returns -2 if the key does not exist.
   * -1 if the key exists but has no associated expire.
   * @see {@link https://redis.io/commands/pttl/}
   */
  async getRateLimitTtl(maxJobs?: number): Promise<number> {
    return this.scripts.getRateLimitTtl(maxJobs);
  }

  /**
   * Get jobId that starts debounced state.
   * @deprecated use getDeduplicationJobId method
   *
   * @param id - debounce identifier
   */
  async getDebounceJobId(id: string): Promise<string | null> {
    const client = await this.client;

    return client.get(`${this.keys.de}:${id}`);
  }

  /**
   * Get jobId from deduplicated state.
   *
   * @param id - deduplication identifier
   */
  async getDeduplicationJobId(id: string): Promise<string | null> {
    const client = await this.client;

    return client.get(`${this.keys.de}:${id}`);
  }
  
  /**
   * Job counts by type
   *
   * Queue#getJobCountByTypes('completed') => completed count
   * Queue#getJobCountByTypes('completed,failed') => completed + failed count
   * Queue#getJobCountByTypes('completed', 'failed') => completed + failed count
   * Queue#getJobCountByTypes('completed', 'waiting', 'failed') => completed + waiting + failed count
   */
  async getJobCountByTypes(...types: JobType[]): Promise<number> {
    const result = await this.getJobCounts(...types);
    return Object.values(result).reduce((sum, count) => sum + count, 0);
  }

  /**
   * Returns the job counts for each type specified or every list/set in the queue by default.
   *
   * @returns An object, key (type) and value (count)
   */
  async getJobCounts(...types: JobType[]): Promise<{
    [index: string]: number;
  }> {
    const currentTypes = this.sanitizeJobTypes(types);

    const responses = await this.scripts.getCounts(currentTypes);

    const counts: { [index: string]: number } = {};
    responses.forEach((res, index) => {
      counts[currentTypes[index]] = res || 0;
    });

    return counts;
  }

  /**
   * Get current job state.
   *
   * @param jobId - job identifier.
   * @returns Returns one of these values:
   * 'completed', 'failed', 'delayed', 'active', 'waiting', 'waiting-children', 'unknown'.
   */
  getJobState(jobId: string): Promise<JobState | 'unknown'> {
    return this.scripts.getState(jobId);
  }

  /**
   * Returns the number of jobs in completed status.
   */
  getCompletedCount(): Promise<number> {
    return this.getJobCountByTypes('completed');
  }

  /**
   * Returns the number of jobs in failed status.
   */
  getFailedCount(): Promise<number> {
    return this.getJobCountByTypes('failed');
  }

  /**
   * Returns the number of jobs in delayed status.
   */
  getDelayedCount(): Promise<number> {
    return this.getJobCountByTypes('delayed');
  }

  /**
   * Returns the number of jobs in active status.
   */
  getActiveCount(): Promise<number> {
    return this.getJobCountByTypes('active');
  }

  /**
   * Returns the number of jobs in prioritized status.
   */
  getPrioritizedCount(): Promise<number> {
    return this.getJobCountByTypes('prioritized');
  }

  /**
   * Returns the number of jobs per priority.
   */
  async getCountsPerPriority(priorities: number[]): Promise<{
    [index: string]: number;
  }> {
    const uniquePriorities = [...new Set(priorities)];
    const responses = await this.scripts.getCountsPerPriority(uniquePriorities);

    const counts: { [index: string]: number } = {};
    responses.forEach((res, index) => {
      counts[`${uniquePriorities[index]}`] = res || 0;
    });

    return counts;
  }

  /**
   * Returns the number of jobs in waiting or paused statuses.
   */
  getWaitingCount(): Promise<number> {
    return this.getJobCountByTypes('waiting');
  }

  /**
   * Returns the number of jobs in waiting-children status.
   */
  getWaitingChildrenCount(): Promise<number> {
    return this.getJobCountByTypes('waiting-children');
  }

  /**
   * Returns the jobs that are in the "waiting" status.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getWaiting(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['waiting'], start, end, true);
  }

  /**
   * Returns the jobs that are in the "waiting-children" status.
   * I.E. parent jobs that have at least one child that has not completed yet.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getWaitingChildren(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['waiting-children'], start, end, true);
  }

  /**
   * Returns the jobs that are in the "active" status.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getActive(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['active'], start, end, true);
  }

  /**
   * Returns the jobs that are in the "delayed" status.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getDelayed(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['delayed'], start, end, true);
  }

  /**
   * Returns the jobs that are in the "prioritized" status.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getPrioritized(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['prioritized'], start, end, true);
  }

  /**
   * Returns the jobs that are in the "completed" status.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getCompleted(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['completed'], start, end, false);
  }

  /**
   * Returns the jobs that are in the "failed" status.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   */
  getFailed(
    start = 0,
    end = -1,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return this.getJobs(['failed'], start, end, false);
  }

  /**
   * Returns the qualified job ids and the raw job data (if available) of the
   * children jobs of the given parent job.
   * It is possible to get either the already processed children, in this case
   * an array of qualified job ids and their result values will be returned,
   * or the pending children, in this case an array of qualified job ids will
   * be returned.
   * A qualified job id is a string representing the job id in a given queue,
   * for example: "bull:myqueue:jobid".
   *
   * @param parentId The id of the parent job
   * @param type "processed" | "pending"
   * @param opts
   *
   * @returns  { items: { id: string, v?: any, err?: string } [], jobs: JobJsonRaw[], total: number}
   */
  async getDependencies(
    parentId: string,
    type: 'processed' | 'pending',
    start: number,
    end: number,
  ): Promise<{
    items: { id: string; v?: any; err?: string }[];
    jobs: JobJsonRaw[];
    total: number;
  }> {
    const key = this.toKey(
      type == 'processed'
        ? `${parentId}:processed`
        : `${parentId}:dependencies`,
    );
    const { items, total, jobs } = await this.scripts.paginate(key, {
      start,
      end,
      fetchJobs: true,
    });
    return {
      items,
      jobs,
      total,
    };
  }

  async getRanges(
    types: JobType[],
    start = 0,
    end = 1,
    asc = false,
  ): Promise<string[]> {
    const multiCommands: string[] = [];

    this.commandByType(types, false, (key, command) => {
      switch (command) {
        case 'lrange':
          multiCommands.push('lrange');
          break;
        case 'zrange':
          multiCommands.push('zrange');
          break;
      }
    });

    const responses = await this.scripts.getRanges(types, start, end, asc);

    let results: string[] = [];

    responses.forEach((response: string[], index: number) => {
      const result = response || [];

      if (asc && multiCommands[index] === 'lrange') {
        results = results.concat(result.reverse());
      } else {
        results = results.concat(result);
      }
    });

    return [...new Set(results)];
  }

  /**
   * Returns the jobs that are on the given statuses (note that JobType is synonym for job status)
   * @param types - the statuses of the jobs to return.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   * @param asc - if true, the jobs will be returned in ascending order.
   */
  async getJobs(
    types?: JobType[] | JobType,
    start = 0,
    end = -1,
    asc = false,
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    const currentTypes = this.sanitizeJobTypes(types);

    const jobIds = await this.getRanges(currentTypes, start, end, asc);

    return Promise.all(
      jobIds.map(
        jobId =>
          this.Job.fromId(this, jobId) as Promise<
            Job<DataType, ResultType, NameType>
          >,
      ),
    );
  }

  /**
   * Returns the logs for a given Job.
   * @param jobId - the id of the job to get the logs for.
   * @param start - zero based index from where to start returning jobs.
   * @param end - zero based index where to stop returning jobs.
   * @param asc - if true, the jobs will be returned in ascending order.
   */
  async getJobLogs(
    jobId: string,
    start = 0,
    end = -1,
    asc = true,
  ): Promise<{ logs: string[]; count: number }> {
    const client = await this.client;
    const multi = client.multi();

    const logsKey = this.toKey(jobId + ':logs');
    if (asc) {
      multi.lrange(logsKey, start, end);
    } else {
      multi.lrange(logsKey, -(end + 1), -(start + 1));
    }
    multi.llen(logsKey);
    const result = (await multi.exec()) as [[Error, [string]], [Error, number]];
    if (!asc) {
      result[0][1].reverse();
    }
    return {
      logs: result[0][1],
      count: result[1][1],
    };
  }

  private async baseGetClients(matcher: (name: string) => boolean): Promise<
    {
      [index: string]: string;
    }[]
  > {
    const client = await this.client;
    try {
      const clients = (await client.client('LIST')) as string;
      const list = this.parseClientList(clients, matcher);
      return list;
    } catch (err) {
      if (!clientCommandMessageReg.test((<Error>err).message)) {
        throw err;
      }

      return [{ name: 'GCP does not support client list' }];
    }
  }

  /**
   * Get the worker list related to the queue. i.e. all the known
   * workers that are available to process jobs for this queue.
   * Note: GCP does not support SETNAME, so this call will not work
   *
   * @returns - Returns an array with workers info.
   */
  getWorkers(): Promise<
    {
      [index: string]: string;
    }[]
  > {
    const unnamedWorkerClientName = `${this.clientName()}`;
    const namedWorkerClientName = `${this.clientName()}:w:`;

    const matcher = (name: string) =>
      name &&
      (name === unnamedWorkerClientName ||
        name.startsWith(namedWorkerClientName));

    return this.baseGetClients(matcher);
  }

  /**
   * Returns the current count of workers for the queue.
   *
   * getWorkersCount(): Promise<number>
   *
   */
  async getWorkersCount(): Promise<number> {
    const workers = await this.getWorkers();
    return workers.length;
  }

  /**
   * Get queue events list related to the queue.
   * Note: GCP does not support SETNAME, so this call will not work
   *
   * @deprecated do not use this method, it will be removed in the future.
   *
   * @returns - Returns an array with queue events info.
   */
  async getQueueEvents(): Promise<
    {
      [index: string]: string;
    }[]
  > {
    const clientName = `${this.clientName()}${QUEUE_EVENT_SUFFIX}`;
    return this.baseGetClients((name: string) => name === clientName);
  }

  /**
   * Get queue metrics related to the queue.
   *
   * This method returns the gathered metrics for the queue.
   * The metrics are represented as an array of job counts
   * per unit of time (1 minute).
   *
   * @param start - Start point of the metrics, where 0
   * is the newest point to be returned.
   * @param end - End point of the metrics, where -1 is the
   * oldest point to be returned.
   *
   * @returns - Returns an object with queue metrics.
   */
  async getMetrics(
    type: 'completed' | 'failed',
    start = 0,
    end = -1,
  ): Promise<Metrics> {
    const client = await this.client;
    const metricsKey = this.toKey(`metrics:${type}`);
    const dataKey = `${metricsKey}:data`;

    const multi = client.multi();
    multi.hmget(metricsKey, 'count', 'prevTS', 'prevCount');
    multi.lrange(dataKey, start, end);
    multi.llen(dataKey);

    const [hmget, range, len] = (await multi.exec()) as [
      [Error, [string, string, string]],
      [Error, []],
      [Error, number],
    ];
    const [err, [count, prevTS, prevCount]] = hmget;
    const [err2, data] = range;
    const [err3, numPoints] = len;
    if (err || err2) {
      throw err || err2 || err3;
    }

    return {
      meta: {
        count: parseInt(count || '0', 10),
        prevTS: parseInt(prevTS || '0', 10),
        prevCount: parseInt(prevCount || '0', 10),
      },
      data,
      count: numPoints,
    };
  }

  private parseClientList(list: string, matcher: (name: string) => boolean) {
    const lines = list.split('\n');
    const clients: { [index: string]: string }[] = [];

    lines.forEach((line: string) => {
      const client: { [index: string]: string } = {};
      const keyValues = line.split(' ');
      keyValues.forEach(function (keyValue) {
        const index = keyValue.indexOf('=');
        const key = keyValue.substring(0, index);
        const value = keyValue.substring(index + 1);
        client[key] = value;
      });
      const name = client['name'];
      if (matcher(name)) {
        client['name'] = this.name;
        client['rawname'] = name;
        clients.push(client);
      }
    });
    return clients;
  }
}

1	/eslint-env node /
2	'use strict';
3
4	import { QueueBase } from './queue-base';	1✔
5	import { Job } from './job';	1✔
6	import { clientCommandMessageReg, QUEUE_EVENT_SUFFIX } from '../utils';	1✔
7	import { JobState, JobType } from '../types';
8	import { JobJsonRaw, Metrics } from '../interfaces';
9
10	/**
11	*
12	* @class QueueGetters
13	* @extends QueueBase
14	*
15	* @description Provides different getters for different aspects of a queue.
16	*/
17	export class QueueGetters<	1✔
18	DataType,
19	ResultType,
20	NameType extends string,
21	> extends QueueBase {
22	getJob(
23	jobId: string,
24	): Promise<Job<DataType, ResultType, NameType> \| undefined> {
25	return this.Job.fromId(this, jobId) as Promise<	201✔
26	Job<DataType, ResultType, NameType>
27	>;
28	}
29
30	private commandByType(
31	types: JobType[],
32	count: boolean,
33	callback: (key: string, dataType: string) => void,
34	) {
35	return types.map((type: string) => {	58✔
36	type = type === 'waiting' ? 'wait' : type; // alias	79✔
37
38	const key = this.toKey(type);	79✔
39
40	switch (type) {	79✔
41	case 'completed':
42	case 'failed':
43	case 'delayed':
44	case 'prioritized':
45	case 'repeat':
46	case 'waiting-children':
47	return callback(key, count ? 'zcard' : 'zrange');	50!
48	case 'active':
49	case 'wait':
50	case 'paused':
51	return callback(key, count ? 'llen' : 'lrange');	29!
52	}
53	});
54	}
55
56	/**
57	* Helper to easily extend Job class calls.
58	*/
59	protected get Job(): typeof Job {
60	return Job;	3,932✔
61	}
62
63	private sanitizeJobTypes(types: JobType[] \| JobType \| undefined): JobType[] {
64	const currentTypes = typeof types === 'string' ? [types] : types;	242✔
65
66	if (Array.isArray(currentTypes) && currentTypes.length > 0) {	242✔
67	const sanitizedTypes = [...currentTypes];	237✔
68
69	if (sanitizedTypes.indexOf('waiting') !== -1) {	237✔
70	sanitizedTypes.push('paused');	78✔
71	}
72
73	return [...new Set(sanitizedTypes)];	237✔
74	}
75
76	return [	5✔
77	'active',
78	'completed',
79	'delayed',
80	'failed',
81	'paused',
82	'prioritized',
83	'waiting',
84	'waiting-children',
85	];
86	}
87
88	/**
89	Returns the number of jobs waiting to be processed. This includes jobs that are
90	"waiting" or "delayed" or "prioritized" or "waiting-children".
91	*/
92	async count(): Promise<number> {
93	const count = await this.getJobCountByTypes(	51✔
94	'waiting',
95	'paused',
96	'delayed',
97	'prioritized',
98	'waiting-children',
99	);
100
101	return count;	51✔
102	}
103
104	/**
105	* Returns the time to live for a rate limited key in milliseconds.
106	* @param maxJobs - max jobs to be considered in rate limit state. If not passed
107	* it will return the remaining ttl without considering if max jobs is excedeed.
108	* @returns -2 if the key does not exist.
109	* -1 if the key exists but has no associated expire.
110	* @see {@link https://redis.io/commands/pttl/}
111	*/
112	async getRateLimitTtl(maxJobs?: number): Promise<number> {
113	return this.scripts.getRateLimitTtl(maxJobs);	35✔
114	}
115
116	/**
117	* Get jobId that starts debounced state.
118	* @deprecated use getDeduplicationJobId method
119	*
120	* @param id - debounce identifier
121	*/
122	async getDebounceJobId(id: string): Promise<string \| null> {
123	const client = await this.client;	2✔
124
125	return client.get(`${this.keys.de}:${id}`);	2✔
126	}
127
128	/**
129	* Get jobId from deduplicated state.
130	*
131	* @param id - deduplication identifier
132	*/
133	async getDeduplicationJobId(id: string): Promise<string \| null> {
NEW 134	const client = await this.client;	×
135
NEW 136	return client.get(`${this.keys.de}:${id}`);	×
137	}
138
139	/**
140	* Job counts by type
141	*
142	* Queue#getJobCountByTypes('completed') => completed count
143	* Queue#getJobCountByTypes('completed,failed') => completed + failed count
144	* Queue#getJobCountByTypes('completed', 'failed') => completed + failed count
145	* Queue#getJobCountByTypes('completed', 'waiting', 'failed') => completed + waiting + failed count
146	*/
147	async getJobCountByTypes(...types: JobType[]): Promise<number> {
148	const result = await this.getJobCounts(...types);	139✔
149	return Object.values(result).reduce((sum, count) => sum + count, 0);	379✔
150	}
151
152	/**
153	* Returns the job counts for each type specified or every list/set in the queue by default.
154	*
155	* @returns An object, key (type) and value (count)
156	*/
157	async getJobCounts(...types: JobType[]): Promise<{
158	[index: string]: number;
159	}> {
160	const currentTypes = this.sanitizeJobTypes(types);	184✔
161
162	const responses = await this.scripts.getCounts(currentTypes);	184✔
163
164	const counts: { [index: string]: number } = {};	184✔
165	responses.forEach((res, index) => {	184✔
166	counts[currentTypes[index]] = res \|\| 0;	447✔
167	});
168
169	return counts;	184✔
170	}
171
172	/**
173	* Get current job state.
174	*
175	* @param jobId - job identifier.
176	* @returns Returns one of these values:
177	* 'completed', 'failed', 'delayed', 'active', 'waiting', 'waiting-children', 'unknown'.
178	*/
179	getJobState(jobId: string): Promise<JobState \| 'unknown'> {
180	return this.scripts.getState(jobId);	3✔
181	}
182
183	/**
184	* Returns the number of jobs in completed status.
185	*/
186	getCompletedCount(): Promise<number> {
187	return this.getJobCountByTypes('completed');	9✔
188	}
189
190	/**
191	* Returns the number of jobs in failed status.
192	*/
193	getFailedCount(): Promise<number> {
194	return this.getJobCountByTypes('failed');	6✔
195	}
196
197	/**
198	* Returns the number of jobs in delayed status.
199	*/
200	getDelayedCount(): Promise<number> {
201	return this.getJobCountByTypes('delayed');	10✔
202	}
203
204	/**
205	* Returns the number of jobs in active status.
206	*/
207	getActiveCount(): Promise<number> {
208	return this.getJobCountByTypes('active');	1✔
209	}
210
211	/**
212	* Returns the number of jobs in prioritized status.
213	*/
214	getPrioritizedCount(): Promise<number> {
215	return this.getJobCountByTypes('prioritized');	1✔
216	}
217
218	/**
219	* Returns the number of jobs per priority.
220	*/
221	async getCountsPerPriority(priorities: number[]): Promise<{
222	[index: string]: number;
223	}> {
224	const uniquePriorities = [...new Set(priorities)];	2✔
225	const responses = await this.scripts.getCountsPerPriority(uniquePriorities);	2✔
226
227	const counts: { [index: string]: number } = {};	2✔
228	responses.forEach((res, index) => {	2✔
229	counts[`${uniquePriorities[index]}`] = res \|\| 0;	8!
230	});
231
232	return counts;	2✔
233	}
234
235	/**
236	* Returns the number of jobs in waiting or paused statuses.
237	*/
238	getWaitingCount(): Promise<number> {
239	return this.getJobCountByTypes('waiting');	5✔
240	}
241
242	/**
243	* Returns the number of jobs in waiting-children status.
244	*/
245	getWaitingChildrenCount(): Promise<number> {
246	return this.getJobCountByTypes('waiting-children');	3✔
247	}
248
249	/**
250	* Returns the jobs that are in the "waiting" status.
251	* @param start - zero based index from where to start returning jobs.
252	* @param end - zero based index where to stop returning jobs.
253	*/
254	getWaiting(
255	start = 0,	7✔
256	end = -1,	7✔
257	): Promise<Job<DataType, ResultType, NameType>[]> {
258	return this.getJobs(['waiting'], start, end, true);	8✔
259	}
260
261	/**
262	* Returns the jobs that are in the "waiting-children" status.
263	* I.E. parent jobs that have at least one child that has not completed yet.
264	* @param start - zero based index from where to start returning jobs.
265	* @param end - zero based index where to stop returning jobs.
266	*/
267	getWaitingChildren(
268	start = 0,	1✔
269	end = -1,	1✔
270	): Promise<Job<DataType, ResultType, NameType>[]> {
271	return this.getJobs(['waiting-children'], start, end, true);	1✔
272	}
273
274	/**
275	* Returns the jobs that are in the "active" status.
276	* @param start - zero based index from where to start returning jobs.
277	* @param end - zero based index where to stop returning jobs.
278	*/
279	getActive(
280	start = 0,	1✔
281	end = -1,	1✔
282	): Promise<Job<DataType, ResultType, NameType>[]> {
283	return this.getJobs(['active'], start, end, true);	1✔
284	}
285
286	/**
287	* Returns the jobs that are in the "delayed" status.
288	* @param start - zero based index from where to start returning jobs.
289	* @param end - zero based index where to stop returning jobs.
290	*/
291	getDelayed(
292	start = 0,	29✔
293	end = -1,	29✔
294	): Promise<Job<DataType, ResultType, NameType>[]> {
295	return this.getJobs(['delayed'], start, end, true);	29✔
296	}
297
298	/**
299	* Returns the jobs that are in the "prioritized" status.
300	* @param start - zero based index from where to start returning jobs.
301	* @param end - zero based index where to stop returning jobs.
302	*/
303	getPrioritized(
304	start = 0,	1✔
305	end = -1,	1✔
306	): Promise<Job<DataType, ResultType, NameType>[]> {
307	return this.getJobs(['prioritized'], start, end, true);	1✔
308	}
309
310	/**
311	* Returns the jobs that are in the "completed" status.
312	* @param start - zero based index from where to start returning jobs.
313	* @param end - zero based index where to stop returning jobs.
314	*/
315	getCompleted(
316	start = 0,	2✔
317	end = -1,	2✔
318	): Promise<Job<DataType, ResultType, NameType>[]> {
319	return this.getJobs(['completed'], start, end, false);	2✔
320	}
321
322	/**
323	* Returns the jobs that are in the "failed" status.
324	* @param start - zero based index from where to start returning jobs.
325	* @param end - zero based index where to stop returning jobs.
326	*/
327	getFailed(
328	start = 0,	3✔
329	end = -1,	3✔
330	): Promise<Job<DataType, ResultType, NameType>[]> {
331	return this.getJobs(['failed'], start, end, false);	4✔
332	}
333
334	/**
335	* Returns the qualified job ids and the raw job data (if available) of the
336	* children jobs of the given parent job.
337	* It is possible to get either the already processed children, in this case
338	* an array of qualified job ids and their result values will be returned,
339	* or the pending children, in this case an array of qualified job ids will
340	* be returned.
341	* A qualified job id is a string representing the job id in a given queue,
342	* for example: "bull:myqueue:jobid".
343	*
344	* @param parentId The id of the parent job
345	* @param type "processed" \| "pending"
346	* @param opts
347	*
348	* @returns { items: { id: string, v?: any, err?: string } [], jobs: JobJsonRaw[], total: number}
349	*/
350	async getDependencies(
351	parentId: string,
352	type: 'processed' \| 'pending',
353	start: number,
354	end: number,
355	): Promise<{
356	items: { id: string; v?: any; err?: string }[];
357	jobs: JobJsonRaw[];
358	total: number;
359	}> {
360	const key = this.toKey(	4✔
361	type == 'processed'	4✔
362	? `${parentId}:processed`
363	: `${parentId}:dependencies`,
364	);
365	const { items, total, jobs } = await this.scripts.paginate(key, {	4✔
366	start,
367	end,
368	fetchJobs: true,
369	});
370	return {	4✔
371	items,
372	jobs,
373	total,
374	};
375	}
376
377	async getRanges(
378	types: JobType[],
379	start = 0,	×
380	end = 1,	×
381	asc = false,	×
382	): Promise<string[]> {
383	const multiCommands: string[] = [];	58✔
384
385	this.commandByType(types, false, (key, command) => {	58✔
386	switch (command) {	79✔
387	case 'lrange':
388	multiCommands.push('lrange');	29✔
389	break;	29✔
390	case 'zrange':
391	multiCommands.push('zrange');	50✔
392	break;	50✔
393	}
394	});
395
396	const responses = await this.scripts.getRanges(types, start, end, asc);	58✔
397
398	let results: string[] = [];	58✔
399
400	responses.forEach((response: string[], index: number) => {	58✔
401	const result = response \|\| [];	79!
402
403	if (asc && multiCommands[index] === 'lrange') {	79✔
404	results = results.concat(result.reverse());	17✔
405	} else {
406	results = results.concat(result);	62✔
407	}
408	});
409
410	return [...new Set(results)];	58✔
411	}
412
413	/**
414	* Returns the jobs that are on the given statuses (note that JobType is synonym for job status)
415	* @param types - the statuses of the jobs to return.
416	* @param start - zero based index from where to start returning jobs.
417	* @param end - zero based index where to stop returning jobs.
418	* @param asc - if true, the jobs will be returned in ascending order.
419	*/
420	async getJobs(
421	types?: JobType[] \| JobType,
422	start = 0,	7✔
423	end = -1,	7✔
424	asc = false,	7✔
425	): Promise<Job<DataType, ResultType, NameType>[]> {
426	const currentTypes = this.sanitizeJobTypes(types);	58✔
427
428	const jobIds = await this.getRanges(currentTypes, start, end, asc);	58✔
429
430	return Promise.all(	58✔
431	jobIds.map(
432	jobId =>
433	this.Job.fromId(this, jobId) as Promise<	88✔
434	Job<DataType, ResultType, NameType>
435	>,
436	),
437	);
438	}
439
440	/**
441	* Returns the logs for a given Job.
442	* @param jobId - the id of the job to get the logs for.
443	* @param start - zero based index from where to start returning jobs.
444	* @param end - zero based index where to stop returning jobs.
445	* @param asc - if true, the jobs will be returned in ascending order.
446	*/
447	async getJobLogs(
448	jobId: string,
449	start = 0,	165✔
450	end = -1,	165✔
451	asc = true,	166✔
452	): Promise<{ logs: string[]; count: number }> {
453	const client = await this.client;	170✔
454	const multi = client.multi();	170✔
455
456	const logsKey = this.toKey(jobId + ':logs');	170✔
457	if (asc) {	170✔
458	multi.lrange(logsKey, start, end);	167✔
459	} else {
460	multi.lrange(logsKey, -(end + 1), -(start + 1));	3✔
461	}
462	multi.llen(logsKey);	170✔
463	const result = (await multi.exec()) as [[Error, [string]], [Error, number]];	170✔
464	if (!asc) {	170✔
465	result[0][1].reverse();	3✔
466	}
467	return {	170✔
468	logs: result[0][1],
469	count: result[1][1],
470	};
471	}
472
473	private async baseGetClients(matcher: (name: string) => boolean): Promise<
474	{
475	[index: string]: string;
476	}[]
477	> {
478	const client = await this.client;	18✔
479	try {	18✔
480	const clients = (await client.client('LIST')) as string;	18✔
481	const list = this.parseClientList(clients, matcher);	18✔
482	return list;	18✔
483	} catch (err) {
484	if (!clientCommandMessageReg.test((<Error>err).message)) {	×
485	throw err;	×
486	}
487
488	return [{ name: 'GCP does not support client list' }];	×
489	}
490	}
491
492	/**
493	* Get the worker list related to the queue. i.e. all the known
494	* workers that are available to process jobs for this queue.
495	* Note: GCP does not support SETNAME, so this call will not work
496	*
497	* @returns - Returns an array with workers info.
498	*/
499	getWorkers(): Promise<
500	{
501	[index: string]: string;
502	}[]
503	> {
504	const unnamedWorkerClientName = `${this.clientName()}`;	16✔
505	const namedWorkerClientName = `${this.clientName()}:w:`;	16✔
506
507	const matcher = (name: string) =>	16✔
508	name &&	250✔
509	(name === unnamedWorkerClientName \|\|
510	name.startsWith(namedWorkerClientName));
511
512	return this.baseGetClients(matcher);	16✔
513	}
514
515	/**
516	* Returns the current count of workers for the queue.
517	*
518	* getWorkersCount(): Promise<number>
519	*
520	*/
521	async getWorkersCount(): Promise<number> {
522	const workers = await this.getWorkers();	5✔
523	return workers.length;	5✔
524	}
525
526	/**
527	* Get queue events list related to the queue.
528	* Note: GCP does not support SETNAME, so this call will not work
529	*
530	* @deprecated do not use this method, it will be removed in the future.
531	*
532	* @returns - Returns an array with queue events info.
533	*/
534	async getQueueEvents(): Promise<
535	{
536	[index: string]: string;
537	}[]
538	> {
539	const clientName = `${this.clientName()}${QUEUE_EVENT_SUFFIX}`;	2✔
540	return this.baseGetClients((name: string) => name === clientName);	31✔
541	}
542
543	/**
544	* Get queue metrics related to the queue.
545	*
546	* This method returns the gathered metrics for the queue.
547	* The metrics are represented as an array of job counts
548	* per unit of time (1 minute).
549	*
550	* @param start - Start point of the metrics, where 0
551	* is the newest point to be returned.
552	* @param end - End point of the metrics, where -1 is the
553	* oldest point to be returned.
554	*
555	* @returns - Returns an object with queue metrics.
556	*/
557	async getMetrics(
558	type: 'completed' \| 'failed',
559	start = 0,	4✔
560	end = -1,	4✔
561	): Promise<Metrics> {
562	const client = await this.client;	11✔
563	const metricsKey = this.toKey(`metrics:${type}`);	11✔
564	const dataKey = `${metricsKey}:data`;	11✔
565
566	const multi = client.multi();	11✔
567	multi.hmget(metricsKey, 'count', 'prevTS', 'prevCount');	11✔
568	multi.lrange(dataKey, start, end);	11✔
569	multi.llen(dataKey);	11✔
570
571	const [hmget, range, len] = (await multi.exec()) as [	11✔
572	[Error, [string, string, string]],
573	[Error, []],
574	[Error, number],
575	];
576	const [err, [count, prevTS, prevCount]] = hmget;	11✔
577	const [err2, data] = range;	11✔
578	const [err3, numPoints] = len;	11✔
579	if (err \|\| err2) {	11!
580	throw err \|\| err2 \|\| err3;	×
581	}
582
583	return {	11✔
584	meta: {
585	count: parseInt(count \|\| '0', 10),	11!
586	prevTS: parseInt(prevTS \|\| '0', 10),	11!
587	prevCount: parseInt(prevCount \|\| '0', 10),	11!
588	},
589	data,
590	count: numPoints,
591	};
592	}
593
594	private parseClientList(list: string, matcher: (name: string) => boolean) {
595	const lines = list.split('\n');	18✔
596	const clients: { [index: string]: string }[] = [];	18✔
597
598	lines.forEach((line: string) => {	18✔
599	const client: { [index: string]: string } = {};	281✔
600	const keyValues = line.split(' ');	281✔
601	keyValues.forEach(function (keyValue) {	281✔
602	const index = keyValue.indexOf('=');	8,171✔
603	const key = keyValue.substring(0, index);	8,171✔
604	const value = keyValue.substring(index + 1);	8,171✔
605	client[key] = value;	8,171✔
606	});
607	const name = client['name'];	281✔
608	if (matcher(name)) {	281✔
609	client['name'] = this.name;	23✔
610	client['rawname'] = name;	23✔
611	clients.push(client);	23✔
612	}
613	});
614	return clients;	18✔
615	}
616	}

taskforcesh / bullmq / 11219305157

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