11272543053

Committed 10 Oct 2024 10:50AM UTC coverage: 88.573% (-0.2%) from 88.819%

Build # 11272543053

Build Type

Pull #2721

github

Committed by

web-flow

Commit Message

Merge 7261e4056 into 07012e6c3

Pull Request Pull Request #2721: feat(queue): observability for queue

Run Details

1366 of 1639 branches covered (83.34%)

Branch coverage included in aggregate %.

286 of 318 new or added lines in 5 files covered. (89.94%)

44 existing lines in 5 files now uncovered.

2494 of 2719 relevant lines covered (91.72%)

4016.75 hits per line

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

85.39

/src/classes/flow-producer.ts

import { EventEmitter } from 'events';
import { Redis, ChainableCommander } from 'ioredis';
import { v4 } from 'uuid';
import {
  FlowJob,
  FlowQueuesOpts,
  FlowOpts,
  IoredisListener,
  QueueBaseOptions,
  RedisClient,
} from '../interfaces';
import { getParentKey, isRedisInstance } from '../utils';
import { Job } from './job';
import { KeysMap, QueueKeys } from './queue-keys';
import { RedisConnection } from './redis-connection';

export interface AddNodeOpts {
  multi: ChainableCommander;
  node: FlowJob;
  parent?: {
    parentOpts: {
      id: string;
      queue: string;
    };
    parentDependenciesKey: string;
  };
  /**
   * Queues options that will be applied in each node depending on queue name presence.
   */
  queuesOpts?: FlowQueuesOpts;
}

export interface AddChildrenOpts {
  multi: ChainableCommander;
  nodes: FlowJob[];
  parent: {
    parentOpts: {
      id: string;
      queue: string;
    };
    parentDependenciesKey: string;
  };
  queuesOpts?: FlowQueuesOpts;
}

export interface NodeOpts {
  /**
   * Root job queue name.
   */
  queueName: string;
  /**
   * Prefix included in job key.
   */
  prefix?: string;
  /**
   * Root job id.
   */
  id: string;
  /**
   * Maximum depth or levels to visit in the tree.
   */
  depth?: number;
  /**
   * Maximum quantity of children per type (processed, unprocessed).
   */
  maxChildren?: number;
}

export interface JobNode {
  job: Job;
  children?: JobNode[];
}

export interface FlowProducerListener extends IoredisListener {
  /**
   * Listen to 'error' event.
   *
   * This event is triggered when an error is throw.
   */
  error: (failedReason: Error) => void;
}

/**
 * This class allows to add jobs with dependencies between them in such
 * a way that it is possible to build complex flows.
 * Note: A flow is a tree-like structure of jobs that depend on each other.
 * Whenever the children of a given parent are completed, the parent
 * will be processed, being able to access the children's result data.
 * All Jobs can be in different queues, either children or parents,
 */
export class FlowProducer extends EventEmitter {
  toKey: (name: string, type: string) => string;
  keys: KeysMap;
  closing: Promise<void> | undefined;
  queueKeys: QueueKeys;

  protected connection: RedisConnection;

  constructor(
    public opts: QueueBaseOptions = { connection: {} },
    Connection: typeof RedisConnection = RedisConnection,
  ) {
    super();

    this.opts = {
      prefix: 'bull',
      ...opts,
    };

    this.connection = new Connection(
      opts.connection,
      isRedisInstance(opts.connection),
      false,
      opts.skipVersionCheck,
    );

    this.connection.on('error', (error: Error) => this.emit('error', error));
    this.connection.on('close', () => {
      if (!this.closing) {
        this.emit('ioredis:close');
      }
    });

    this.queueKeys = new QueueKeys(opts.prefix);
  }

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

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

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

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

  /**
   * Returns a promise that resolves to a redis client. Normally used only by subclasses.
   */
  get client(): Promise<RedisClient> {
    return this.connection.client;
  }

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

  waitUntilReady(): Promise<RedisClient> {
    return this.client;
  }

  /**
   * Adds a flow.
   *
   * This call would be atomic, either it fails and no jobs will
   * be added to the queues, or it succeeds and all jobs will be added.
   *
   * @param flow - an object with a tree-like structure where children jobs
   * will be processed before their parents.
   * @param opts - options that will be applied to the flow object.
   */
  async add(flow: FlowJob, opts?: FlowOpts): Promise<JobNode> {
    if (this.closing) {
      return;
    }
    const client = await this.connection.client;
    const multi = client.multi();

    const parentOpts = flow?.opts?.parent;
    const parentKey = getParentKey(parentOpts);
    const parentDependenciesKey = parentKey
      ? `${parentKey}:dependencies`
      : undefined;

    const jobsTree = this.addNode({
      multi,
      node: flow,
      queuesOpts: opts?.queuesOptions,
      parent: {
        parentOpts,
        parentDependenciesKey,
      },
    });

    await multi.exec();

    return jobsTree;
  }

  /**
   * Get a flow.
   *
   * @param opts - an object with options for getting a JobNode.
   */
  async getFlow(opts: NodeOpts): Promise<JobNode> {
    if (this.closing) {
      return;
    }
    const client = await this.connection.client;

    const updatedOpts = Object.assign(
      {
        depth: 10,
        maxChildren: 20,
      },
      opts,
    );
    const jobsTree = this.getNode(client, updatedOpts);

    return jobsTree;
  }

  /**
   * Adds multiple flows.
   *
   * A flow is a tree-like structure of jobs that depend on each other.
   * Whenever the children of a given parent are completed, the parent
   * will be processed, being able to access the children's result data.
   *
   * All Jobs can be in different queues, either children or parents,
   * however this call would be atomic, either it fails and no jobs will
   * be added to the queues, or it succeeds and all jobs will be added.
   *
   * @param flows - an array of objects with a tree-like structure where children jobs
   * will be processed before their parents.
   */
  async addBulk(flows: FlowJob[]): Promise<JobNode[]> {
    if (this.closing) {
      return;
    }
    const client = await this.connection.client;
    const multi = client.multi();

    const jobsTrees = this.addNodes(multi, flows);

    await multi.exec();

    return jobsTrees;
  }

  /**
   * Add a node (job) of a flow to the queue. This method will recursively
   * add all its children as well. Note that a given job can potentially be
   * a parent and a child job at the same time depending on where it is located
   * in the tree hierarchy.
   *
   * @param multi - ioredis ChainableCommander
   * @param node - the node representing a job to be added to some queue
   * @param parent - parent data sent to children to create the "links" to their parent
   * @returns
   */
  protected addNode({ multi, node, parent, queuesOpts }: AddNodeOpts): JobNode {
    const prefix = node.prefix || this.opts.prefix;
    const queue = this.queueFromNode(node, new QueueKeys(prefix), prefix);
    const queueOpts = queuesOpts && queuesOpts[node.queueName];

    const jobsOpts = queueOpts?.defaultJobOptions ?? {};
    const jobId = node.opts?.jobId || v4();

    const job = new this.Job(
      queue,
      node.name,
      node.data,
      {
        ...jobsOpts,
        ...node.opts,
        parent: parent?.parentOpts,
      },
      jobId,
    );

    const parentKey = getParentKey(parent?.parentOpts);

    if (node.children && node.children.length > 0) {
      // Create parent job, will be a job in status "waiting-children".
      const parentId = jobId;
      const queueKeysParent = new QueueKeys(node.prefix || this.opts.prefix);
      const waitChildrenKey = queueKeysParent.toKey(
        node.queueName,
        'waiting-children',
      );

      job.addJob(<Redis>(multi as unknown), {
        parentDependenciesKey: parent?.parentDependenciesKey,
        waitChildrenKey,
        parentKey,
      });

      const parentDependenciesKey = `${queueKeysParent.toKey(
        node.queueName,
        parentId,
      )}:dependencies`;

      const children = this.addChildren({
        multi,
        nodes: node.children,
        parent: {
          parentOpts: {
            id: parentId,
            queue: queueKeysParent.getQueueQualifiedName(node.queueName),
          },
          parentDependenciesKey,
        },
        queuesOpts,
      });

      return { job, children };
    } else {
      job.addJob(<Redis>(multi as unknown), {
        parentDependenciesKey: parent?.parentDependenciesKey,
        parentKey,
      });

      return { job };
    }
  }

  /**
   * Adds nodes (jobs) of multiple flows to the queue. This method will recursively
   * add all its children as well. Note that a given job can potentially be
   * a parent and a child job at the same time depending on where it is located
   * in the tree hierarchy.
   *
   * @param multi - ioredis ChainableCommander
   * @param nodes - the nodes representing jobs to be added to some queue
   * @returns
   */
  protected addNodes(multi: ChainableCommander, nodes: FlowJob[]): JobNode[] {
    return nodes.map(node => {
      const parentOpts = node?.opts?.parent;
      const parentKey = getParentKey(parentOpts);
      const parentDependenciesKey = parentKey
        ? `${parentKey}:dependencies`
        : undefined;

      return this.addNode({
        multi,
        node,
        parent: {
          parentOpts,
          parentDependenciesKey,
        },
      });
    });
  }

  private async getNode(client: RedisClient, node: NodeOpts): Promise<JobNode> {
    const queue = this.queueFromNode(
      node,
      new QueueKeys(node.prefix),
      node.prefix,
    );

    const job = await this.Job.fromId(queue, node.id);

    if (job) {
      const { processed = {}, unprocessed = [] } = await job.getDependencies({
        processed: {
          count: node.maxChildren,
        },
        unprocessed: {
          count: node.maxChildren,
        },
      });
      const processedKeys = Object.keys(processed);

      const childrenCount = processedKeys.length + unprocessed.length;
      const newDepth = node.depth - 1;
      if (childrenCount > 0 && newDepth) {
        const children = await this.getChildren(
          client,
          [...processedKeys, ...unprocessed],
          newDepth,
          node.maxChildren,
        );

        return { job, children };
      } else {
        return { job };
      }
    }
  }

  private addChildren({ multi, nodes, parent, queuesOpts }: AddChildrenOpts) {
    return nodes.map(node => this.addNode({ multi, node, parent, queuesOpts }));
  }

  private getChildren(
    client: RedisClient,
    childrenKeys: string[],
    depth: number,
    maxChildren: number,
  ) {
    const getChild = (key: string) => {
      const [prefix, queueName, id] = key.split(':');

      return this.getNode(client, {
        id,
        queueName,
        prefix,
        depth,
        maxChildren,
      });
    };

    return Promise.all([...childrenKeys.map(getChild)]);
  }

  /**
   * Helper factory method that creates a queue-like object
   * required to create jobs in any queue.
   *
   * @param node -
   * @param queueKeys -
   * @returns
   */
  private queueFromNode(
    node: Omit<NodeOpts, 'id' | 'depth' | 'maxChildren'>,
    queueKeys: QueueKeys,
    prefix: string,
  ) {
    return {
      client: this.connection.client,
      name: node.queueName,
      keys: queueKeys.getKeys(node.queueName),
      toKey: (type: string) => queueKeys.toKey(node.queueName, type),
      opts: { prefix, connection: {} },
      qualifiedName: queueKeys.getQueueQualifiedName(node.queueName),
      closing: this.closing,
      waitUntilReady: async () => this.connection.client,
      removeListener: this.removeListener.bind(this) as any,
      emit: this.emit.bind(this) as any,
      on: this.on.bind(this) as any,
      redisVersion: this.connection.redisVersion,
    };
  }

  /**
   *
   * Closes the connection and returns a promise that resolves when the connection is closed.
   */
  async close(): Promise<void> {
    if (!this.closing) {
      this.closing = this.connection.close();
    }
    await this.closing;
  }

  /**
   *
   * Force disconnects a connection.
   */
  disconnect(): Promise<void> {
    return this.connection.disconnect();
  }
}

1	import { EventEmitter } from 'events';	1✔
2	import { Redis, ChainableCommander } from 'ioredis';
3	import { v4 } from 'uuid';	1✔
4	import {
5	FlowJob,
6	FlowQueuesOpts,
7	FlowOpts,
8	IoredisListener,
9	QueueBaseOptions,
10	RedisClient,
11	} from '../interfaces';
12	import { getParentKey, isRedisInstance } from '../utils';	1✔
13	import { Job } from './job';	1✔
14	import { KeysMap, QueueKeys } from './queue-keys';	1✔
15	import { RedisConnection } from './redis-connection';	1✔
16
17	export interface AddNodeOpts {
18	multi: ChainableCommander;
19	node: FlowJob;
20	parent?: {
21	parentOpts: {
22	id: string;
23	queue: string;
24	};
25	parentDependenciesKey: string;
26	};
27	/**
28	* Queues options that will be applied in each node depending on queue name presence.
29	*/
30	queuesOpts?: FlowQueuesOpts;
31	}
32
33	export interface AddChildrenOpts {
34	multi: ChainableCommander;
35	nodes: FlowJob[];
36	parent: {
37	parentOpts: {
38	id: string;
39	queue: string;
40	};
41	parentDependenciesKey: string;
42	};
43	queuesOpts?: FlowQueuesOpts;
44	}
45
46	export interface NodeOpts {
47	/**
48	* Root job queue name.
49	*/
50	queueName: string;
51	/**
52	* Prefix included in job key.
53	*/
54	prefix?: string;
55	/**
56	* Root job id.
57	*/
58	id: string;
59	/**
60	* Maximum depth or levels to visit in the tree.
61	*/
62	depth?: number;
63	/**
64	* Maximum quantity of children per type (processed, unprocessed).
65	*/
66	maxChildren?: number;
67	}
68
69	export interface JobNode {
70	job: Job;
71	children?: JobNode[];
72	}
73
74	export interface FlowProducerListener extends IoredisListener {
75	/**
76	* Listen to 'error' event.
77	*
78	* This event is triggered when an error is throw.
79	*/
80	error: (failedReason: Error) => void;
81	}
82
83	/**
84	* This class allows to add jobs with dependencies between them in such
85	* a way that it is possible to build complex flows.
86	* Note: A flow is a tree-like structure of jobs that depend on each other.
87	* Whenever the children of a given parent are completed, the parent
88	* will be processed, being able to access the children's result data.
89	* All Jobs can be in different queues, either children or parents,
90	*/
91	export class FlowProducer extends EventEmitter {	1✔
92	toKey: (name: string, type: string) => string;
93	keys: KeysMap;
94	closing: Promise<void> \| undefined;
95	queueKeys: QueueKeys;
96
97	protected connection: RedisConnection;
98
99	constructor(
100	public opts: QueueBaseOptions = { connection: {} },	74!
101	Connection: typeof RedisConnection = RedisConnection,	74✔
102	) {
103	super();	74✔
104
105	this.opts = {	74✔
106	prefix: 'bull',
107	...opts,
108	};
109
110	this.connection = new Connection(	74✔
111	opts.connection,
112	isRedisInstance(opts.connection),
113	false,
114	opts.skipVersionCheck,
115	);
116
117	this.connection.on('error', (error: Error) => this.emit('error', error));	74✔
118	this.connection.on('close', () => {	74✔
119	if (!this.closing) {	1!
120	this.emit('ioredis:close');	1✔
121	}
122	});
123
124	this.queueKeys = new QueueKeys(opts.prefix);	74✔
125	}
126
127	emit<U extends keyof FlowProducerListener>(
128	event: U,
129	...args: Parameters<FlowProducerListener[U]>
130	): boolean {
131	return super.emit(event, ...args);	12✔
132	}
133
134	off<U extends keyof FlowProducerListener>(
135	eventName: U,
136	listener: FlowProducerListener[U],
137	): this {
138	super.off(eventName, listener);	×
139	return this;	×
140	}
141
142	on<U extends keyof FlowProducerListener>(
143	event: U,
144	listener: FlowProducerListener[U],
145	): this {
146	super.on(event, listener);	×
147	return this;	×
148	}
149
150	once<U extends keyof FlowProducerListener>(
151	event: U,
152	listener: FlowProducerListener[U],
153	): this {
154	super.once(event, listener);	×
155	return this;	×
156	}
157
158	/**
159	* Returns a promise that resolves to a redis client. Normally used only by subclasses.
160	*/
161	get client(): Promise<RedisClient> {
162	return this.connection.client;	1✔
163	}
164
165	/**
166	* Helper to easily extend Job class calls.
167	*/
168	protected get Job(): typeof Job {
169	return Job;	341✔
170	}
171
172	waitUntilReady(): Promise<RedisClient> {
173	return this.client;	×
174	}
175
176	/**
177	* Adds a flow.
178	*
179	* This call would be atomic, either it fails and no jobs will
180	* be added to the queues, or it succeeds and all jobs will be added.
181	*
182	* @param flow - an object with a tree-like structure where children jobs
183	* will be processed before their parents.
184	* @param opts - options that will be applied to the flow object.
185	*/
186	async add(flow: FlowJob, opts?: FlowOpts): Promise<JobNode> {
187	if (this.closing) {	76!
188	return;	×
189	}
190	const client = await this.connection.client;	76✔
191	const multi = client.multi();	76✔
192
193	const parentOpts = flow?.opts?.parent;	76!
194	const parentKey = getParentKey(parentOpts);	76✔
195	const parentDependenciesKey = parentKey	76✔
196	? `${parentKey}:dependencies`
197	: undefined;
198
199	const jobsTree = this.addNode({	76✔
200	multi,
201	node: flow,
202	queuesOpts: opts?.queuesOptions,	228✔
203	parent: {
204	parentOpts,
205	parentDependenciesKey,
206	},
207	});
208
209	await multi.exec();	76✔
210
211	return jobsTree;	76✔
212	}
213
214	/**
215	* Get a flow.
216	*
217	* @param opts - an object with options for getting a JobNode.
218	*/
219	async getFlow(opts: NodeOpts): Promise<JobNode> {
220	if (this.closing) {	2!
221	return;	×
222	}
223	const client = await this.connection.client;	2✔
224
225	const updatedOpts = Object.assign(	2✔
226	{
227	depth: 10,
228	maxChildren: 20,
229	},
230	opts,
231	);
232	const jobsTree = this.getNode(client, updatedOpts);	2✔
233
234	return jobsTree;	2✔
235	}
236
237	/**
238	* Adds multiple flows.
239	*
240	* A flow is a tree-like structure of jobs that depend on each other.
241	* Whenever the children of a given parent are completed, the parent
242	* will be processed, being able to access the children's result data.
243	*
244	* All Jobs can be in different queues, either children or parents,
245	* however this call would be atomic, either it fails and no jobs will
246	* be added to the queues, or it succeeds and all jobs will be added.
247	*
248	* @param flows - an array of objects with a tree-like structure where children jobs
249	* will be processed before their parents.
250	*/
251	async addBulk(flows: FlowJob[]): Promise<JobNode[]> {
252	if (this.closing) {	3!
253	return;	×
254	}
255	const client = await this.connection.client;	3✔
256	const multi = client.multi();	3✔
257
258	const jobsTrees = this.addNodes(multi, flows);	3✔
259
260	await multi.exec();	3✔
261
262	return jobsTrees;	3✔
263	}
264
265	/**
266	* Add a node (job) of a flow to the queue. This method will recursively
267	* add all its children as well. Note that a given job can potentially be
268	* a parent and a child job at the same time depending on where it is located
269	* in the tree hierarchy.
270	*
271	* @param multi - ioredis ChainableCommander
272	* @param node - the node representing a job to be added to some queue
273	* @param parent - parent data sent to children to create the "links" to their parent
274	* @returns
275	*/
276	protected addNode({ multi, node, parent, queuesOpts }: AddNodeOpts): JobNode {
277	const prefix = node.prefix \|\| this.opts.prefix;	334✔
278	const queue = this.queueFromNode(node, new QueueKeys(prefix), prefix);	334✔
279	const queueOpts = queuesOpts && queuesOpts[node.queueName];	334✔
280
281	const jobsOpts = queueOpts?.defaultJobOptions ?? {};	334✔
282	const jobId = node.opts?.jobId \|\| v4();	334✔
283
284	const job = new this.Job(	334✔
285	queue,
286	node.name,
287	node.data,
288	{
289	...jobsOpts,
290	...node.opts,
291	parent: parent?.parentOpts,	1,002!
292	},
293	jobId,
294	);
295
296	const parentKey = getParentKey(parent?.parentOpts);	334!
297
298	if (node.children && node.children.length > 0) {	334✔
299	// Create parent job, will be a job in status "waiting-children".
300	const parentId = jobId;	106✔
301	const queueKeysParent = new QueueKeys(node.prefix \|\| this.opts.prefix);	106✔
302	const waitChildrenKey = queueKeysParent.toKey(	106✔
303	node.queueName,
304	'waiting-children',
305	);
306
307	job.addJob(<Redis>(multi as unknown), {	106✔
308	parentDependenciesKey: parent?.parentDependenciesKey,	318!
309	waitChildrenKey,
310	parentKey,
311	});
312
313	const parentDependenciesKey = `${queueKeysParent.toKey(	106✔
314	node.queueName,
315	parentId,
316	)}:dependencies`;
317
318	const children = this.addChildren({	106✔
319	multi,
320	nodes: node.children,
321	parent: {
322	parentOpts: {
323	id: parentId,
324	queue: queueKeysParent.getQueueQualifiedName(node.queueName),
325	},
326	parentDependenciesKey,
327	},
328	queuesOpts,
329	});
330
331	return { job, children };	106✔
332	} else {
333	job.addJob(<Redis>(multi as unknown), {	228✔
334	parentDependenciesKey: parent?.parentDependenciesKey,	684!
335	parentKey,
336	});
337
338	return { job };	228✔
339	}
340	}
341
342	/**
343	* Adds nodes (jobs) of multiple flows to the queue. This method will recursively
344	* add all its children as well. Note that a given job can potentially be
345	* a parent and a child job at the same time depending on where it is located
346	* in the tree hierarchy.
347	*
348	* @param multi - ioredis ChainableCommander
349	* @param nodes - the nodes representing jobs to be added to some queue
350	* @returns
351	*/
352	protected addNodes(multi: ChainableCommander, nodes: FlowJob[]): JobNode[] {
353	return nodes.map(node => {	3✔
354	const parentOpts = node?.opts?.parent;	8!
355	const parentKey = getParentKey(parentOpts);	8✔
356	const parentDependenciesKey = parentKey	8✔
357	? `${parentKey}:dependencies`
358	: undefined;
359
360	return this.addNode({	8✔
361	multi,
362	node,
363	parent: {
364	parentOpts,
365	parentDependenciesKey,
366	},
367	});
368	});
369	}
370
371	private async getNode(client: RedisClient, node: NodeOpts): Promise<JobNode> {
372	const queue = this.queueFromNode(	7✔
373	node,
374	new QueueKeys(node.prefix),
375	node.prefix,
376	);
377
378	const job = await this.Job.fromId(queue, node.id);	7✔
379
380	if (job) {	7!
381	const { processed = {}, unprocessed = [] } = await job.getDependencies({	7!
382	processed: {
383	count: node.maxChildren,
384	},
385	unprocessed: {
386	count: node.maxChildren,
387	},
388	});
389	const processedKeys = Object.keys(processed);	7✔
390
391	const childrenCount = processedKeys.length + unprocessed.length;	7✔
392	const newDepth = node.depth - 1;	7✔
393	if (childrenCount > 0 && newDepth) {	7✔
394	const children = await this.getChildren(	4✔
395	client,
396	[...processedKeys, ...unprocessed],
397	newDepth,
398	node.maxChildren,
399	);
400
401	return { job, children };	4✔
402	} else {
403	return { job };	3✔
404	}
405	}
406	}
407
408	private addChildren({ multi, nodes, parent, queuesOpts }: AddChildrenOpts) {
409	return nodes.map(node => this.addNode({ multi, node, parent, queuesOpts }));	250✔
410	}
411
412	private getChildren(
413	client: RedisClient,
414	childrenKeys: string[],
415	depth: number,
416	maxChildren: number,
417	) {
418	const getChild = (key: string) => {	4✔
419	const [prefix, queueName, id] = key.split(':');	5✔
420
421	return this.getNode(client, {	5✔
422	id,
423	queueName,
424	prefix,
425	depth,
426	maxChildren,
427	});
428	};
429
430	return Promise.all([...childrenKeys.map(getChild)]);	4✔
431	}
432
433	/**
434	* Helper factory method that creates a queue-like object
435	* required to create jobs in any queue.
436	*
437	* @param node -
438	* @param queueKeys -
439	* @returns
440	*/
441	private queueFromNode(
442	node: Omit<NodeOpts, 'id' \| 'depth' \| 'maxChildren'>,
443	queueKeys: QueueKeys,
444	prefix: string,
445	) {
446	return {	341✔
447	client: this.connection.client,
448	name: node.queueName,
449	keys: queueKeys.getKeys(node.queueName),
450	toKey: (type: string) => queueKeys.toKey(node.queueName, type),	636✔
451	opts: { prefix, connection: {} },
452	qualifiedName: queueKeys.getQueueQualifiedName(node.queueName),
453	closing: this.closing,
454	waitUntilReady: async () => this.connection.client,	12✔
455	removeListener: this.removeListener.bind(this) as any,
456	emit: this.emit.bind(this) as any,
457	on: this.on.bind(this) as any,
458	redisVersion: this.connection.redisVersion,
459	};
460	}
461
462	/**
463	*
464	* Closes the connection and returns a promise that resolves when the connection is closed.
465	*/
466	async close(): Promise<void> {
467	if (!this.closing) {	73!
468	this.closing = this.connection.close();	73✔
469	}
470	await this.closing;	73✔
471	}
472
473	/**
474	*
475	* Force disconnects a connection.
476	*/
477	disconnect(): Promise<void> {
UNCOV 478	return this.connection.disconnect();	×
479	}
480	}

taskforcesh / bullmq / 11272543053

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