7070091515

Committed 02 Dec 2023 12:12PM UTC coverage: 89.495% (-0.05%) from 89.543%

Build # 7070091515

Build Type

Pull #2309

github

Committed by

web-flow

Commit Message

Merge 0bac0fbb9 into 9896b8a70

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

Run Details

1016 of 1220 branches covered (0.0%)

Branch coverage included in aggregate %.

30 of 32 new or added lines in 7 files covered. (93.75%)

59 existing lines in 7 files now uncovered.

2034 of 2188 relevant lines covered (92.96%)

3037.13 hits per line

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

84.93

/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 '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?: typeof RedisConnection,
  ) {
    super(
      name,
      {
        ...opts,
        connection: isRedisInstance(connection)
          ? (<RedisClient>connection).duplicate()
          : connection,
        blockingConnection: true,
      },
      Connection,
    );

    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;

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

taskforcesh / bullmq / 7070091515

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