11452721079

Committed 22 Oct 2024 03:31AM UTC coverage: 88.381% (+0.02%) from 88.364%

Build # 11452721079

Build Type

Pull #2851

github

Committed by

web-flow

Commit Message

Merge 801b8a122 into d1eb1bba5

Pull Request Pull Request #2851: fix(connection): do not allow to set blockingConnection option

Run Details

1273 of 1539 branches covered (82.72%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

6 existing lines in 1 file now uncovered.

2401 of 2618 relevant lines covered (91.71%)

4259.0 hits per line

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

85.53

/src/classes/queue-events.ts

import {
  IoredisListener,
  QueueEventsOptions,
  RedisClient,
  StreamReadRaw,
} from '../interfaces';
import {
  array2obj,
  clientCommandMessageReg,
  isRedisInstance,
  QUEUE_EVENT_SUFFIX,
} from '../utils';
import { QueueBase } from './queue-base';
import { RedisConnection } from './redis-connection';

export interface QueueEventsListener extends IoredisListener {
  /**
   * Listen to 'active' event.
   *
   * This event is triggered when a job enters the 'active' state.
   */
  active: (args: { jobId: string; prev?: string }, id: string) => void;

  /**
   * Listen to 'added' event.
   *
   * This event is triggered when a job is created.
   */
  added: (args: { jobId: string; name: string }, id: string) => void;

  /**
   * Listen to 'cleaned' event.
   *
   * This event is triggered when a cleaned method is triggered.
   */
  cleaned: (args: { count: string }, id: string) => void;

  /**
   * Listen to 'completed' event.
   *
   * This event is triggered when a job has successfully completed.
   */
  completed: (
    args: { jobId: string; returnvalue: string; prev?: string },
    id: string,
  ) => void;

  /**
   * Listen to 'debounced' event.
   * @deprecated use deduplicated event
   *
   * This event is triggered when a job is debounced because debounceId still existed.
   */
  debounced: (args: { jobId: string; debounceId: string }, id: string) => void;

  /**
   * Listen to 'deduplicated' event.
   *
   * This event is triggered when a job is deduplicated because deduplicatedId still existed.
   */
  deduplicated: (
    args: { jobId: string; deduplicationId: string },
    id: string,
  ) => void;

  /**
   * Listen to 'delayed' event.
   *
   * This event is triggered when a job is delayed.
   */
  delayed: (args: { jobId: string; delay: number }, id: string) => void;

  /**
   * Listen to 'drained' event.
   *
   * This event is triggered when the queue has drained the waiting list.
   * Note that there could still be delayed jobs waiting their timers to expire
   * and this event will still be triggered as long as the waiting list has emptied.
   */
  drained: (id: string) => void;

  /**
   * Listen to 'duplicated' event.
   *
   * This event is triggered when a job is not created because it already exist.
   */
  duplicated: (args: { jobId: string }, id: string) => void;

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

  /**
   * Listen to 'failed' event.
   *
   * This event is triggered when a job has thrown an exception.
   */
  failed: (
    args: { jobId: string; failedReason: string; prev?: string },
    id: string,
  ) => void;

  /**
   * Listen to 'paused' event.
   *
   * This event is triggered when a queue is paused.
   */
  paused: (args: {}, id: string) => void;

  /**
   * Listen to 'progress' event.
   *
   * This event is triggered when a job updates it progress, i.e. the
   * Job##updateProgress() method is called. This is useful to notify
   * progress or any other data from within a processor to the rest of the
   * world.
   */
  progress: (
    args: { jobId: string; data: number | object },
    id: string,
  ) => void;

  /**
   * Listen to 'removed' event.
   *
   * This event is triggered when a job has been manually
   * removed from the queue.
   */
  removed: (args: { jobId: string; prev: string }, id: string) => void;

  /**
   * Listen to 'resumed' event.
   *
   * This event is triggered when a queue is resumed.
   */
  resumed: (args: {}, id: string) => void;

  /**
   * Listen to 'retries-exhausted' event.
   *
   * This event is triggered when a job has retried the maximum attempts.
   */
  'retries-exhausted': (
    args: { jobId: string; attemptsMade: string },
    id: string,
  ) => void;

  /**
   * Listen to 'stalled' event.
   *
   * This event is triggered when a job has been moved from 'active' back
   * to 'waiting'/'failed' due to the processor not being able to renew
   * the lock on the said job.
   */
  stalled: (args: { jobId: string }, id: string) => void;

  /**
   * Listen to 'waiting' event.
   *
   * This event is triggered when a job enters the 'waiting' state.
   */
  waiting: (args: { jobId: string; prev?: string }, id: string) => void;

  /**
   * Listen to 'waiting-children' event.
   *
   * This event is triggered when a job enters the 'waiting-children' state.
   */
  'waiting-children': (args: { jobId: string }, id: string) => void;
}

/**
 * The QueueEvents class is used for listening to the global events
 * emitted by a given queue.
 *
 * This class requires a dedicated redis connection.
 *
 */
export class QueueEvents extends QueueBase {
  private running = false;

  constructor(
    name: string,
    { connection, autorun = true, ...opts }: QueueEventsOptions = {
      connection: {},
    },
    Connection?: typeof RedisConnection,
  ) {
    super(
      name,
      {
        ...opts,
        connection: isRedisInstance(connection)
          ? (<RedisClient>connection).duplicate()
          : connection,
      },
      Connection,
      true,
    );

    this.opts = Object.assign(
      {
        blockingTimeout: 10000,
      },
      this.opts,
    );

    if (autorun) {
      this.run().catch(error => this.emit('error', error));
    }
  }

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

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

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

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

  /**
   * Manually starts running the event consumming loop. This shall be used if you do not
   * use the default "autorun" option on the constructor.
   */
  async run(): Promise<void> {
    if (!this.running) {
      try {
        this.running = true;
        const client = await this.client;

        // TODO: Planed for deprecation as it has no really a use case
        try {
          await client.client('SETNAME', this.clientName(QUEUE_EVENT_SUFFIX));
        } catch (err) {
          if (!clientCommandMessageReg.test((<Error>err).message)) {
            throw err;
          }
        }

        await this.consumeEvents(client);
      } catch (error) {
        this.running = false;
        throw error;
      }
    } else {
      throw new Error('Queue Events is already running.');
    }
  }

  private async consumeEvents(client: RedisClient): Promise<void> {
    const opts: QueueEventsOptions = this.opts;

    const key = this.keys.events;
    let id = opts.lastEventId || '$';

    while (!this.closing) {
      // Cast to actual return type, see: https://github.com/DefinitelyTyped/DefinitelyTyped/issues/44301
      const data: StreamReadRaw = await this.checkConnectionError(() =>
        client.xread('BLOCK', opts.blockingTimeout!, 'STREAMS', key, id),
      );
      if (data) {
        const stream = data[0];
        const events = stream[1];

        for (let i = 0; i < events.length; i++) {
          id = events[i][0];
          const args = array2obj(events[i][1]);

          //
          // TODO: we may need to have a separate xtream for progress data
          // to avoid this hack.
          switch (args.event) {
            case 'progress':
              args.data = JSON.parse(args.data);
              break;
            case 'completed':
              args.returnvalue = JSON.parse(args.returnvalue);
              break;
          }

          const { event, ...restArgs } = args;

          if (event === 'drained') {
            this.emit(event, id);
          } else {
            this.emit(event as any, restArgs, id);
            if (restArgs.jobId) {
              this.emit(`${event}:${restArgs.jobId}` as any, restArgs, id);
            }
          }
        }
      }
    }
  }

  /**
   * Stops consuming events and close the underlying Redis connection if necessary.
   *
   * @returns
   */
  close(): Promise<void> {
    if (!this.closing) {
      this.closing = this.disconnect();
    }
    return this.closing;
  }
}

1	import {
2	IoredisListener,
3	QueueEventsOptions,
4	RedisClient,
5	StreamReadRaw,
6	} from '../interfaces';
7	import {	1✔
8	array2obj,
9	clientCommandMessageReg,
10	isRedisInstance,
11	QUEUE_EVENT_SUFFIX,
12	} from '../utils';
13	import { QueueBase } from './queue-base';	1✔
14	import { RedisConnection } from './redis-connection';
15
16	export interface QueueEventsListener extends IoredisListener {
17	/**
18	* Listen to 'active' event.
19	*
20	* This event is triggered when a job enters the 'active' state.
21	*/
22	active: (args: { jobId: string; prev?: string }, id: string) => void;
23
24	/**
25	* Listen to 'added' event.
26	*
27	* This event is triggered when a job is created.
28	*/
29	added: (args: { jobId: string; name: string }, id: string) => void;
30
31	/**
32	* Listen to 'cleaned' event.
33	*
34	* This event is triggered when a cleaned method is triggered.
35	*/
36	cleaned: (args: { count: string }, id: string) => void;
37
38	/**
39	* Listen to 'completed' event.
40	*
41	* This event is triggered when a job has successfully completed.
42	*/
43	completed: (
44	args: { jobId: string; returnvalue: string; prev?: string },
45	id: string,
46	) => void;
47
48	/**
49	* Listen to 'debounced' event.
50	* @deprecated use deduplicated event
51	*
52	* This event is triggered when a job is debounced because debounceId still existed.
53	*/
54	debounced: (args: { jobId: string; debounceId: string }, id: string) => void;
55
56	/**
57	* Listen to 'deduplicated' event.
58	*
59	* This event is triggered when a job is deduplicated because deduplicatedId still existed.
60	*/
61	deduplicated: (
62	args: { jobId: string; deduplicationId: string },
63	id: string,
64	) => void;
65
66	/**
67	* Listen to 'delayed' event.
68	*
69	* This event is triggered when a job is delayed.
70	*/
71	delayed: (args: { jobId: string; delay: number }, id: string) => void;
72
73	/**
74	* Listen to 'drained' event.
75	*
76	* This event is triggered when the queue has drained the waiting list.
77	* Note that there could still be delayed jobs waiting their timers to expire
78	* and this event will still be triggered as long as the waiting list has emptied.
79	*/
80	drained: (id: string) => void;
81
82	/**
83	* Listen to 'duplicated' event.
84	*
85	* This event is triggered when a job is not created because it already exist.
86	*/
87	duplicated: (args: { jobId: string }, id: string) => void;
88
89	/**
90	* Listen to 'error' event.
91	*
92	* This event is triggered when an exception is thrown.
93	*/
94	error: (args: Error) => void;
95
96	/**
97	* Listen to 'failed' event.
98	*
99	* This event is triggered when a job has thrown an exception.
100	*/
101	failed: (
102	args: { jobId: string; failedReason: string; prev?: string },
103	id: string,
104	) => void;
105
106	/**
107	* Listen to 'paused' event.
108	*
109	* This event is triggered when a queue is paused.
110	*/
111	paused: (args: {}, id: string) => void;
112
113	/**
114	* Listen to 'progress' event.
115	*
116	* This event is triggered when a job updates it progress, i.e. the
117	* Job##updateProgress() method is called. This is useful to notify
118	* progress or any other data from within a processor to the rest of the
119	* world.
120	*/
121	progress: (
122	args: { jobId: string; data: number \| object },
123	id: string,
124	) => void;
125
126	/**
127	* Listen to 'removed' event.
128	*
129	* This event is triggered when a job has been manually
130	* removed from the queue.
131	*/
132	removed: (args: { jobId: string; prev: string }, id: string) => void;
133
134	/**
135	* Listen to 'resumed' event.
136	*
137	* This event is triggered when a queue is resumed.
138	*/
139	resumed: (args: {}, id: string) => void;
140
141	/**
142	* Listen to 'retries-exhausted' event.
143	*
144	* This event is triggered when a job has retried the maximum attempts.
145	*/
146	'retries-exhausted': (
147	args: { jobId: string; attemptsMade: string },
148	id: string,
149	) => void;
150
151	/**
152	* Listen to 'stalled' event.
153	*
154	* This event is triggered when a job has been moved from 'active' back
155	* to 'waiting'/'failed' due to the processor not being able to renew
156	* the lock on the said job.
157	*/
158	stalled: (args: { jobId: string }, id: string) => void;
159
160	/**
161	* Listen to 'waiting' event.
162	*
163	* This event is triggered when a job enters the 'waiting' state.
164	*/
165	waiting: (args: { jobId: string; prev?: string }, id: string) => void;
166
167	/**
168	* Listen to 'waiting-children' event.
169	*
170	* This event is triggered when a job enters the 'waiting-children' state.
171	*/
172	'waiting-children': (args: { jobId: string }, id: string) => void;
173	}
174
175	/**
176	* The QueueEvents class is used for listening to the global events
177	* emitted by a given queue.
178	*
179	* This class requires a dedicated redis connection.
180	*
181	*/
182	export class QueueEvents extends QueueBase {	1✔
183	private running = false;	375✔
184
185	constructor(
186	name: string,
187	{ connection, autorun = true, ...opts }: QueueEventsOptions = {	375!
188	connection: {},
189	},
190	Connection?: typeof RedisConnection,
191	) {
192	super(	375✔
193	name,
194	{
195	...opts,
196	connection: isRedisInstance(connection)	375!
197	? (<RedisClient>connection).duplicate()
198	: connection,
199	},
200	Connection,
201	true,
202	);
203
204	this.opts = Object.assign(	375✔
205	{
206	blockingTimeout: 10000,
207	},
208	this.opts,
209	);
210
211	if (autorun) {	375✔
212	this.run().catch(error => this.emit('error', error));	373✔
213	}
214	}
215
216	emit<U extends keyof QueueEventsListener>(
217	event: U,
218	...args: Parameters<QueueEventsListener[U]>
219	): boolean {
220	return super.emit(event, ...args);	29,088✔
221	}
222
223	off<U extends keyof QueueEventsListener>(
224	eventName: U,
225	listener: QueueEventsListener[U],
226	): this {
UNCOV 227	super.off(eventName, listener);	×
UNCOV 228	return this;	×
229	}
230
231	on<U extends keyof QueueEventsListener>(
232	event: U,
233	listener: QueueEventsListener[U],
234	): this {
235	super.on(event, listener);	140✔
236	return this;	140✔
237	}
238
239	once<U extends keyof QueueEventsListener>(
240	event: U,
241	listener: QueueEventsListener[U],
242	): this {
243	super.once(event, listener);	14✔
244	return this;	14✔
245	}
246
247	/**
248	* Manually starts running the event consumming loop. This shall be used if you do not
249	* use the default "autorun" option on the constructor.
250	*/
251	async run(): Promise<void> {
252	if (!this.running) {	376✔
253	try {	375✔
254	this.running = true;	375✔
255	const client = await this.client;	375✔
256
257	// TODO: Planed for deprecation as it has no really a use case
258	try {	375✔
259	await client.client('SETNAME', this.clientName(QUEUE_EVENT_SUFFIX));	375✔
260	} catch (err) {
UNCOV 261	if (!clientCommandMessageReg.test((<Error>err).message)) {	×
UNCOV 262	throw err;	×
263	}
264	}
265
266	await this.consumeEvents(client);	375✔
267	} catch (error) {
UNCOV 268	this.running = false;	×
UNCOV 269	throw error;	×
270	}
271	} else {
272	throw new Error('Queue Events is already running.');	1✔
273	}
274	}
275
276	private async consumeEvents(client: RedisClient): Promise<void> {
277	const opts: QueueEventsOptions = this.opts;	375✔
278
279	const key = this.keys.events;	375✔
280	let id = opts.lastEventId \|\| '$';	375✔
281
282	while (!this.closing) {	375✔
283	// Cast to actual return type, see: https://github.com/DefinitelyTyped/DefinitelyTyped/issues/44301
284	const data: StreamReadRaw = await this.checkConnectionError(() =>	5,180✔
285	client.xread('BLOCK', opts.blockingTimeout!, 'STREAMS', key, id),	5,180✔
286	);
287	if (data) {	5,179✔
288	const stream = data[0];	4,788✔
289	const events = stream[1];	4,788✔
290
291	for (let i = 0; i < events.length; i++) {	4,788✔
292	id = events[i][0];	14,683✔
293	const args = array2obj(events[i][1]);	14,683✔
294
295	//
296	// TODO: we may need to have a separate xtream for progress data
297	// to avoid this hack.
298	switch (args.event) {	14,683✔
299	case 'progress':
300	args.data = JSON.parse(args.data);	510✔
301	break;	510✔
302	case 'completed':
303	args.returnvalue = JSON.parse(args.returnvalue);	1,848✔
304	break;	1,848✔
305	}
306
307	const { event, ...restArgs } = args;	14,683✔
308
309	if (event === 'drained') {	14,683✔
310	this.emit(event, id);	227✔
311	} else {
312	this.emit(event as any, restArgs, id);	14,456✔
313	if (restArgs.jobId) {	14,456✔
314	this.emit(`${event}:${restArgs.jobId}` as any, restArgs, id);	14,405✔
315	}
316	}
317	}
318	}
319	}
320	}
321
322	/**
323	* Stops consuming events and close the underlying Redis connection if necessary.
324	*
325	* @returns
326	*/
327	close(): Promise<void> {
328	if (!this.closing) {	374!
329	this.closing = this.disconnect();	374✔
330	}
331	return this.closing;	374✔
332	}
333	}

taskforcesh / bullmq / 11452721079

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