11225197208

Committed 07 Oct 2024 10:45PM UTC coverage: 88.744% (-0.06%) from 88.802%

Build # 11225197208

Build Type

Pull #2807

github

Committed by

web-flow

Commit Message

Merge 3e9a351fa into 592938a7f

Pull Request Pull Request #2807: fix(repeat): return current time when immediately and every options

Run Details

1253 of 1507 branches covered (83.15%)

Branch coverage included in aggregate %.

18 of 20 new or added lines in 4 files covered. (90.0%)

1 existing line in 1 file now uncovered.

2366 of 2571 relevant lines covered (92.03%)

4203.83 hits per line

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

72.96

/src/utils.ts

import { parseExpression } from 'cron-parser';
import { Cluster, Redis } from 'ioredis';

// Note: this Polyfill is only needed for Node versions < 15.4.0
import { AbortController } from 'node-abort-controller';

// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore
import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';
import { ChildMessage, RedisClient, RepeatOptions } from './interfaces';
import { EventEmitter } from 'events';
import * as semver from 'semver';
import { ErrorCode } from './enums';

export const errorObject: { [index: string]: any } = { value: null };

export function tryCatch(
  fn: (...args: any) => any,
  ctx: any,
  args: any[],
): any {
  try {
    return fn.apply(ctx, args);
  } catch (e) {
    errorObject.value = e;
    return errorObject;
  }
}

/**
 * Checks the size of string for ascii/non-ascii characters
 * @see https://stackoverflow.com/a/23318053/1347170
 * @param str -
 */
export function lengthInUtf8Bytes(str: string): number {
  return Buffer.byteLength(str, 'utf8');
}

export function isEmpty(obj: object): boolean {
  for (const key in obj) {
    if (Object.prototype.hasOwnProperty.call(obj, key)) {
      return false;
    }
  }
  return true;
}

export function array2obj(arr: string[]): Record<string, string> {
  const obj: { [index: string]: string } = {};
  for (let i = 0; i < arr.length; i += 2) {
    obj[arr[i]] = arr[i + 1];
  }
  return obj;
}

export function delay(
  ms: number,
  abortController?: AbortController,
): Promise<void> {
  return new Promise(resolve => {
    let timeout: ReturnType<typeof setTimeout> | undefined;
    const callback = () => {
      abortController?.signal.removeEventListener('abort', callback);
      clearTimeout(timeout);
      resolve();
    };
    timeout = setTimeout(callback, ms);
    abortController?.signal.addEventListener('abort', callback);
  });
}

export function increaseMaxListeners(
  emitter: EventEmitter,
  count: number,
): void {
  const maxListeners = emitter.getMaxListeners();
  emitter.setMaxListeners(maxListeners + count);
}

export const invertObject = (obj: Record<string, string>) => {
  return Object.entries(obj).reduce<Record<string, string>>(
    (encodeMap, [key, value]) => {
      encodeMap[value] = key;
      return encodeMap;
    },
    {},
  );
};

export function isRedisInstance(obj: any): obj is Redis | Cluster {
  if (!obj) {
    return false;
  }
  const redisApi = ['connect', 'disconnect', 'duplicate'];
  return redisApi.every(name => typeof obj[name] === 'function');
}

export function isRedisCluster(obj: unknown): obj is Cluster {
  return isRedisInstance(obj) && (<Cluster>obj).isCluster;
}

export function decreaseMaxListeners(
  emitter: EventEmitter,
  count: number,
): void {
  increaseMaxListeners(emitter, -count);
}

export async function removeAllQueueData(
  client: RedisClient,
  queueName: string,
  prefix = process.env.BULLMQ_TEST_PREFIX || 'bull',
): Promise<void | boolean> {
  if (client instanceof Cluster) {
    // todo compat with cluster ?
    // @see https://github.com/luin/ioredis/issues/175
    return Promise.resolve(false);
  }
  const pattern = `${prefix}:${queueName}:*`;
  const removing = await new Promise<void>((resolve, reject) => {
    const stream = client.scanStream({
      match: pattern,
    });
    stream.on('data', (keys: string[]) => {
      if (keys.length) {
        const pipeline = client.pipeline();
        keys.forEach(key => {
          pipeline.del(key);
        });
        pipeline.exec().catch(error => {
          reject(error);
        });
      }
    });
    stream.on('end', () => resolve());
    stream.on('error', error => reject(error));
  });
  await removing;
  await client.quit();
}

export function getParentKey(opts: {
  id: string;
  queue: string;
}): string | undefined {
  if (opts) {
    return `${opts.queue}:${opts.id}`;
  }
}

export const clientCommandMessageReg =
  /ERR unknown command ['`]\s*client\s*['`]/;

export const DELAY_TIME_5 = 5000;

export const DELAY_TIME_1 = 100;

export function isNotConnectionError(error: Error): boolean {
  const errorMessage = `${(error as Error).message}`;
  return (
    errorMessage !== CONNECTION_CLOSED_ERROR_MSG &&
    !errorMessage.includes('ECONNREFUSED')
  );
}

interface procSendLike {
  send?(message: any, callback?: (error: Error | null) => void): boolean;
  postMessage?(message: any): void;
}

export const asyncSend = <T extends procSendLike>(
  proc: T,
  msg: any,
): Promise<void> => {
  return new Promise((resolve, reject) => {
    if (typeof proc.send === 'function') {
      proc.send(msg, (err: Error | null) => {
        if (err) {
          reject(err);
        } else {
          resolve();
        }
      });
    } else if (typeof proc.postMessage === 'function') {
      resolve(proc.postMessage(msg));
    } else {
      resolve();
    }
  });
};

export const childSend = (
  proc: NodeJS.Process,
  msg: ChildMessage,
): Promise<void> => asyncSend<NodeJS.Process>(proc, msg);

export const isRedisVersionLowerThan = (
  currentVersion: string,
  minimumVersion: string,
): boolean => {
  const version = semver.valid(semver.coerce(currentVersion)) as string;

  return semver.lt(version, minimumVersion);
};

export const parseObjectValues = (obj: {
  [key: string]: string;
}): Record<string, any> => {
  const accumulator: Record<string, any> = {};
  for (const value of Object.entries(obj)) {
    accumulator[value[0]] = JSON.parse(value[1]);
  }

  return accumulator;
};

export const errorToJSON = (value: any): Record<string, any> => {
  const error: Record<string, any> = {};

  Object.getOwnPropertyNames(value).forEach(function (propName: string) {
    error[propName] = value[propName];
  });

  return error;
};

const INFINITY = 1 / 0;

export const toString = (value: any): string => {
  if (value == null) {
    return '';
  }
  // Exit early for strings to avoid a performance hit in some environments.
  if (typeof value === 'string') {
    return value;
  }
  if (Array.isArray(value)) {
    // Recursively convert values (susceptible to call stack limits).
    return `${value.map(other => (other == null ? other : toString(other)))}`;
  }
  if (
    typeof value == 'symbol' ||
    Object.prototype.toString.call(value) == '[object Symbol]'
  ) {
    return value.toString();
  }
  const result = `${value}`;
  return result === '0' && 1 / value === -INFINITY ? '-0' : result;
};

export const QUEUE_EVENT_SUFFIX = ':qe';

export const finishedErrors = ({
  code,
  jobId,
  parentKey,
  command,
  state,
}: {
  code: number;
  jobId?: string;
  parentKey?: string;
  command: string;
  state?: string;
}): Error => {
  switch (code) {
    case ErrorCode.JobNotExist:
      return new Error(`Missing key for job ${jobId}. ${command}`);
    case ErrorCode.JobLockNotExist:
      return new Error(`Missing lock for job ${jobId}. ${command}`);
    case ErrorCode.JobNotInState:
      return new Error(`Job ${jobId} is not in the ${state} state. ${command}`);
    case ErrorCode.JobPendingDependencies:
      return new Error(`Job ${jobId} has pending dependencies. ${command}`);
    case ErrorCode.ParentJobNotExist:
      return new Error(`Missing key for parent job ${parentKey}. ${command}`);
    case ErrorCode.JobLockMismatch:
      return new Error(
        `Lock mismatch for job ${jobId}. Cmd ${command} from ${state}`,
      );
    case ErrorCode.ParentJobCannotBeReplaced:
      return new Error(
        `The parent job ${parentKey} cannot be replaced. ${command}`,
      );
    case ErrorCode.JobBelongsToJobScheduler:
      return new Error(
        `Job ${jobId} belongs to a job scheduler and cannot be removed directly`,
      );
    default:
      return new Error(`Unknown code ${code} error for ${jobId}. ${command}`);
  }
};

export const getNextMillis = (
  millis: number,
  opts: RepeatOptions,
): number | undefined => {
  const { every, pattern } = opts;

  if (opts.immediately) {
    return new Date().getTime();
  }

  if (every) {
    return Math.floor(millis / opts.every) * opts.every + opts.every;
  }

  const currentDate =
    opts.startDate && new Date(opts.startDate) > new Date(millis)
      ? new Date(opts.startDate)
      : new Date(millis);
  const interval = parseExpression(pattern, {
    ...opts,
    currentDate,
  });

  try {
    return interval.next().getTime();
  } catch (e) {
    // Ignore error
  }
};

1	import { parseExpression } from 'cron-parser';	1✔
2	import { Cluster, Redis } from 'ioredis';	1✔
3
4	// Note: this Polyfill is only needed for Node versions < 15.4.0
5	import { AbortController } from 'node-abort-controller';
6
7	// eslint-disable-next-line @typescript-eslint/ban-ts-comment
8	// @ts-ignore
9	import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';	1✔
10	import { ChildMessage, RedisClient, RepeatOptions } from './interfaces';
11	import { EventEmitter } from 'events';
12	import * as semver from 'semver';	1✔
13	import { ErrorCode } from './enums';	1✔
14
15	export const errorObject: { [index: string]: any } = { value: null };	1✔
16
17	export function tryCatch(	1✔
18	fn: (...args: any) => any,
19	ctx: any,
20	args: any[],
21	): any {
22	try {	5,756✔
23	return fn.apply(ctx, args);	5,756✔
24	} catch (e) {
25	errorObject.value = e;	3,265✔
26	return errorObject;	3,265✔
27	}
28	}
29
30	/**
31	* Checks the size of string for ascii/non-ascii characters
32	* @see https://stackoverflow.com/a/23318053/1347170
33	* @param str -
34	*/
35	export function lengthInUtf8Bytes(str: string): number {	1✔
36	return Buffer.byteLength(str, 'utf8');	4✔
37	}
38
39	export function isEmpty(obj: object): boolean {	1✔
40	for (const key in obj) {	359✔
41	if (Object.prototype.hasOwnProperty.call(obj, key)) {	206!
42	return false;	206✔
43	}
44	}
45	return true;	153✔
46	}
47
48	export function array2obj(arr: string[]): Record<string, string> {	1✔
49	const obj: { [index: string]: string } = {};	17,813✔
50	for (let i = 0; i < arr.length; i += 2) {	17,813✔
51	obj[arr[i]] = arr[i + 1];	68,062✔
52	}
53	return obj;	17,813✔
54	}
55
56	export function delay(	1✔
57	ms: number,
58	abortController?: AbortController,
59	): Promise<void> {
60	return new Promise(resolve => {	1,575✔
61	let timeout: ReturnType<typeof setTimeout> \| undefined;
62	const callback = () => {	1,575✔
63	abortController?.signal.removeEventListener('abort', callback);	1,575✔
64	clearTimeout(timeout);	1,575✔
65	resolve();	1,575✔
66	};
67	timeout = setTimeout(callback, ms);	1,575✔
68	abortController?.signal.addEventListener('abort', callback);	1,575✔
69	});
70	}
71
72	export function increaseMaxListeners(	1✔
73	emitter: EventEmitter,
74	count: number,
75	): void {
76	const maxListeners = emitter.getMaxListeners();	6,503✔
77	emitter.setMaxListeners(maxListeners + count);	6,503✔
78	}
79
80	export const invertObject = (obj: Record<string, string>) => {	1✔
81	return Object.entries(obj).reduce<Record<string, string>>(	1✔
82	(encodeMap, [key, value]) => {
83	encodeMap[value] = key;	5✔
84	return encodeMap;	5✔
85	},
86	{},
87	);
88	};
89
90	export function isRedisInstance(obj: any): obj is Redis \| Cluster {	1✔
91	if (!obj) {	7,095✔
92	return false;	2✔
93	}
94	const redisApi = ['connect', 'disconnect', 'duplicate'];	7,093✔
95	return redisApi.every(name => typeof obj[name] === 'function');	21,239✔
96	}
97
98	export function isRedisCluster(obj: unknown): obj is Cluster {	1✔
99	return isRedisInstance(obj) && (<Cluster>obj).isCluster;	2,232✔
100	}
101
102	export function decreaseMaxListeners(	1✔
103	emitter: EventEmitter,
104	count: number,
105	): void {
106	increaseMaxListeners(emitter, -count);	3,003✔
107	}
108
109	export async function removeAllQueueData(	1✔
110	client: RedisClient,
111	queueName: string,
112	prefix = process.env.BULLMQ_TEST_PREFIX \|\| 'bull',	1,914✔
113	): Promise<void \| boolean> {
114	if (client instanceof Cluster) {	640!
115	// todo compat with cluster ?
116	// @see https://github.com/luin/ioredis/issues/175
117	return Promise.resolve(false);	×
118	}
119	const pattern = `${prefix}:${queueName}:*`;	640✔
120	const removing = await new Promise<void>((resolve, reject) => {	640✔
121	const stream = client.scanStream({	640✔
122	match: pattern,
123	});
124	stream.on('data', (keys: string[]) => {	640✔
125	if (keys.length) {	3,097✔
126	const pipeline = client.pipeline();	2,211✔
127	keys.forEach(key => {	2,211✔
128	pipeline.del(key);	6,093✔
129	});
130	pipeline.exec().catch(error => {	2,211✔
131	reject(error);	×
132	});
133	}
134	});
135	stream.on('end', () => resolve());	640✔
136	stream.on('error', error => reject(error));	640✔
137	});
138	await removing;	640✔
139	await client.quit();	640✔
140	}
141
142	export function getParentKey(opts: {	1✔
143	id: string;
144	queue: string;
145	}): string \| undefined {
146	if (opts) {	17,644✔
147	return `${opts.queue}:${opts.id}`;	880✔
148	}
149	}
150
151	export const clientCommandMessageReg =	1✔
152	/ERR unknown command ['`]\sclient\s['`]/;
153
154	export const DELAY_TIME_5 = 5000;	1✔
155
156	export const DELAY_TIME_1 = 100;	1✔
157
158	export function isNotConnectionError(error: Error): boolean {	1✔
159	const errorMessage = `${(error as Error).message}`;	605✔
160	return (	605✔
161	errorMessage !== CONNECTION_CLOSED_ERROR_MSG &&	605!
162	!errorMessage.includes('ECONNREFUSED')
163	);
164	}
165
166	interface procSendLike {
167	send?(message: any, callback?: (error: Error \| null) => void): boolean;
168	postMessage?(message: any): void;
169	}
170
171	export const asyncSend = <T extends procSendLike>(	1✔
172	proc: T,
173	msg: any,
174	): Promise<void> => {
175	return new Promise((resolve, reject) => {	×
176	if (typeof proc.send === 'function') {	×
177	proc.send(msg, (err: Error \| null) => {	×
178	if (err) {	×
179	reject(err);	×
180	} else {
181	resolve();	×
182	}
183	});
184	} else if (typeof proc.postMessage === 'function') {	×
185	resolve(proc.postMessage(msg));	×
186	} else {
187	resolve();	×
188	}
189	});
190	};
191
192	export const childSend = (	1✔
193	proc: NodeJS.Process,
194	msg: ChildMessage,
195	): Promise<void> => asyncSend<NodeJS.Process>(proc, msg);	×
196
197	export const isRedisVersionLowerThan = (	1✔
198	currentVersion: string,
199	minimumVersion: string,
200	): boolean => {
201	const version = semver.valid(semver.coerce(currentVersion)) as string;	9,038✔
202
203	return semver.lt(version, minimumVersion);	9,038✔
204	};
205
206	export const parseObjectValues = (obj: {	1✔
207	[key: string]: string;
208	}): Record<string, any> => {
209	const accumulator: Record<string, any> = {};	28✔
210	for (const value of Object.entries(obj)) {	28✔
211	accumulator[value[0]] = JSON.parse(value[1]);	41✔
212	}
213
214	return accumulator;	28✔
215	};
216
217	export const errorToJSON = (value: any): Record<string, any> => {	1✔
218	const error: Record<string, any> = {};	×
219
220	Object.getOwnPropertyNames(value).forEach(function (propName: string) {	×
221	error[propName] = value[propName];	×
222	});
223
224	return error;	×
225	};
226
227	const INFINITY = 1 / 0;	1✔
228
229	export const toString = (value: any): string => {	1✔
230	if (value == null) {	×
231	return '';	×
232	}
233	// Exit early for strings to avoid a performance hit in some environments.
234	if (typeof value === 'string') {	×
235	return value;	×
236	}
237	if (Array.isArray(value)) {	×
238	// Recursively convert values (susceptible to call stack limits).
239	return `${value.map(other => (other == null ? other : toString(other)))}`;	×
240	}
241	if (	×
242	typeof value == 'symbol' \|\|	×
243	Object.prototype.toString.call(value) == '[object Symbol]'
244	) {
245	return value.toString();	×
246	}
247	const result = `${value}`;	×
248	return result === '0' && 1 / value === -INFINITY ? '-0' : result;	×
249	};
250
251	export const QUEUE_EVENT_SUFFIX = ':qe';	1✔
252
253	export const finishedErrors = ({	1✔
254	code,
255	jobId,
256	parentKey,
257	command,
258	state,
259	}: {
260	code: number;
261	jobId?: string;
262	parentKey?: string;
263	command: string;
264	state?: string;
265	}): Error => {
266	switch (code) {	33!
267	case ErrorCode.JobNotExist:
268	return new Error(`Missing key for job ${jobId}. ${command}`);	9✔
269	case ErrorCode.JobLockNotExist:
270	return new Error(`Missing lock for job ${jobId}. ${command}`);	7✔
271	case ErrorCode.JobNotInState:
272	return new Error(`Job ${jobId} is not in the ${state} state. ${command}`);	7✔
273	case ErrorCode.JobPendingDependencies:
274	return new Error(`Job ${jobId} has pending dependencies. ${command}`);	1✔
275	case ErrorCode.ParentJobNotExist:
276	return new Error(`Missing key for parent job ${parentKey}. ${command}`);	2✔
277	case ErrorCode.JobLockMismatch:
278	return new Error(	2✔
279	`Lock mismatch for job ${jobId}. Cmd ${command} from ${state}`,
280	);
281	case ErrorCode.ParentJobCannotBeReplaced:
282	return new Error(	1✔
283	`The parent job ${parentKey} cannot be replaced. ${command}`,
284	);
285	case ErrorCode.JobBelongsToJobScheduler:
286	return new Error(	4✔
287	`Job ${jobId} belongs to a job scheduler and cannot be removed directly`,
288	);
289	default:
UNCOV 290	return new Error(`Unknown code ${code} error for ${jobId}. ${command}`);	×
291	}
292	};
293
294	export const getNextMillis = (	1✔
295	millis: number,
296	opts: RepeatOptions,
297	): number \| undefined => {
298	const { every, pattern } = opts;	361✔
299
300	if (opts.immediately) {	361✔
301	return new Date().getTime();	8✔
302	}
303
304	if (every) {	353✔
305	return Math.floor(millis / opts.every) * opts.every + opts.every;	114✔
306	}
307
308	const currentDate =
309	opts.startDate && new Date(opts.startDate) > new Date(millis)	239✔
310	? new Date(opts.startDate)
311	: new Date(millis);
312	const interval = parseExpression(pattern, {	239✔
313	...opts,
314	currentDate,
315	});
316
317	try {	239✔
318	return interval.next().getTime();	239✔
319	} catch (e) {
320	// Ignore error
321	}
322	};

taskforcesh / bullmq / 11225197208

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