7095970812

Committed 05 Dec 2023 03:57AM UTC coverage: 89.572% (+0.01%) from 89.561%

Build # 7095970812

Build Type

Pull #2309

github

Committed by

web-flow

Commit Message

Merge 84015ffa0 into 30d082c8f

Pull Request Pull Request #2309: Feat/better handling of queue markers

Run Details

1018 of 1221 branches covered (0.0%)

Branch coverage included in aggregate %.

19 of 20 new or added lines in 3 files covered. (95.0%)

58 existing lines in 7 files now uncovered.

2040 of 2193 relevant lines covered (93.02%)

3031.59 hits per line

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

84.97

/src/classes/flow-producer.ts

import { EventEmitter } from 'events';
import { get } from 'lodash';
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>;
  queueKeys: QueueKeys;

  protected connection: RedisConnection;

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

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

    this.connection = new Connection(
      opts.connection,
      opts.blockingConnection,
      opts.skipVersionCheck,
    );
    this.connection = new Connection(
      opts.connection,
      isRedisInstance(opts?.connection),
      false,
      opts.skipVersionCheck,
    );
    this.connection.on('error', error => this.emit('error', error));
    this.connection.on('close', this.emit.bind(this, '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;
  }

  /**
   * 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;
  }

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

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

  /**
   * 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 = get(queueOpts, 'defaultJobOptions');
    const jobId = node.opts?.jobId || v4();

    const job = new this.Job(
      queue,
      node.name,
      node.data,
      {
        ...(jobsOpts ? 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 },
      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,
    };
  }

  close(): Promise<void> {
    if (!this.closing) {
      this.closing = this.connection.close();
    }
    return this.closing;
  }

  disconnect(): Promise<void> {
    return this.connection.disconnect();
  }
}

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

taskforcesh / bullmq / 7095970812

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