11473539716

Committed 23 Oct 2024 05:18AM UTC coverage: 88.491% (+0.01%) from 88.478%

Build # 11473539716

Build Type

Pull #2844

github

Committed by

web-flow

Commit Message

Merge 03f409b93 into a615c6d69

Pull Request Pull Request #2844: feat(queue-events): add QueueEventsProducer for publishing custom events

Run Details

1274 of 1540 branches covered (82.73%)

Branch coverage included in aggregate %.

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

1 existing line in 1 file now uncovered.

2409 of 2622 relevant lines covered (91.88%)

4218.86 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;
}

type CustomParameters<T> = T extends (...args: infer Args) => void
  ? Args
  : never;

type KeyOf<T extends object> = Extract<keyof T, string>;

/**
 * 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,
        blockingConnection: true,
      },
      Connection,
    );

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

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

  emit<
    QEL extends QueueEventsListener = QueueEventsListener,
    U extends KeyOf<QEL> = KeyOf<QEL>,
  >(event: U, ...args: CustomParameters<QEL[U]>): boolean {
    return super.emit(event, ...args);
  }

  off<
    QEL extends QueueEventsListener = QueueEventsListener,
    U extends KeyOf<QEL> = KeyOf<QEL>,
  >(eventName: U, listener: QEL[U]): this {
    super.off(eventName, listener as (...args: any[]) => void);
    return this;
  }

  on<
    QEL extends QueueEventsListener = QueueEventsListener,
    U extends KeyOf<QEL> = KeyOf<QEL>,
  >(event: U, listener: QEL[U]): this {
    super.on(event, listener as (...args: any[]) => void);
    return this;
  }

  once<
    QEL extends QueueEventsListener = QueueEventsListener,
    U extends KeyOf<QEL> = KeyOf<QEL>,
  >(event: U, listener: QEL[U]): this {
    super.once(event, listener as (...args: any[]) => void);
    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	type CustomParameters<T> = T extends (...args: infer Args) => void
176	? Args
177	: never;
178
179	type KeyOf<T extends object> = Extract<keyof T, string>;
180
181	/**
182	* The QueueEvents class is used for listening to the global events
183	* emitted by a given queue.
184	*
185	* This class requires a dedicated redis connection.
186	*
187	*/
188	export class QueueEvents extends QueueBase {	1✔
189	private running = false;	377✔
190
191	constructor(
192	name: string,
193	{ connection, autorun = true, ...opts }: QueueEventsOptions = {	377!
194	connection: {},
195	},
196	Connection?: typeof RedisConnection,
197	) {
198	super(	377✔
199	name,
200	{
201	...opts,
202	connection: isRedisInstance(connection)	377!
203	? (<RedisClient>connection).duplicate()
204	: connection,
205	blockingConnection: true,
206	},
207	Connection,
208	);
209
210	this.opts = Object.assign(	377✔
211	{
212	blockingTimeout: 10000,
213	},
214	this.opts,
215	);
216
217	if (autorun) {	377✔
218	this.run().catch(error => this.emit('error', error));	374✔
219	}
220	}
221
222	emit<
223	QEL extends QueueEventsListener = QueueEventsListener,
224	U extends KeyOf<QEL> = KeyOf<QEL>,
225	>(event: U, ...args: CustomParameters<QEL[U]>): boolean {
226	return super.emit(event, ...args);	29,017✔
227	}
228
229	off<
230	QEL extends QueueEventsListener = QueueEventsListener,
231	U extends KeyOf<QEL> = KeyOf<QEL>,
232	>(eventName: U, listener: QEL[U]): this {
NEW 233	super.off(eventName, listener as (...args: any[]) => void);	×
UNCOV 234	return this;	×
235	}
236
237	on<
238	QEL extends QueueEventsListener = QueueEventsListener,
239	U extends KeyOf<QEL> = KeyOf<QEL>,
240	>(event: U, listener: QEL[U]): this {
241	super.on(event, listener as (...args: any[]) => void);	141✔
242	return this;	141✔
243	}
244
245	once<
246	QEL extends QueueEventsListener = QueueEventsListener,
247	U extends KeyOf<QEL> = KeyOf<QEL>,
248	>(event: U, listener: QEL[U]): this {
249	super.once(event, listener as (...args: any[]) => void);	14✔
250	return this;	14✔
251	}
252
253	/**
254	* Manually starts running the event consumming loop. This shall be used if you do not
255	* use the default "autorun" option on the constructor.
256	*/
257	async run(): Promise<void> {
258	if (!this.running) {	378✔
259	try {	377✔
260	this.running = true;	377✔
261	const client = await this.client;	377✔
262
263	// TODO: Planed for deprecation as it has no really a use case
264	try {	377✔
265	await client.client('SETNAME', this.clientName(QUEUE_EVENT_SUFFIX));	377✔
266	} catch (err) {
267	if (!clientCommandMessageReg.test((<Error>err).message)) {	×
268	throw err;	×
269	}
270	}
271
272	await this.consumeEvents(client);	377✔
273	} catch (error) {
274	this.running = false;	×
275	throw error;	×
276	}
277	} else {
278	throw new Error('Queue Events is already running.');	1✔
279	}
280	}
281
282	private async consumeEvents(client: RedisClient): Promise<void> {
283	const opts: QueueEventsOptions = this.opts;	377✔
284
285	const key = this.keys.events;	377✔
286	let id = opts.lastEventId \|\| '$';	377✔
287
288	while (!this.closing) {	377✔
289	// Cast to actual return type, see: https://github.com/DefinitelyTyped/DefinitelyTyped/issues/44301
290	const data: StreamReadRaw = await this.checkConnectionError(() =>	5,177✔
291	client.xread('BLOCK', opts.blockingTimeout!, 'STREAMS', key, id),	5,177✔
292	);
293	if (data) {	5,176✔
294	const stream = data[0];	4,783✔
295	const events = stream[1];	4,783✔
296
297	for (let i = 0; i < events.length; i++) {	4,783✔
298	id = events[i][0];	14,648✔
299	const args = array2obj(events[i][1]);	14,648✔
300
301	//
302	// TODO: we may need to have a separate xtream for progress data
303	// to avoid this hack.
304	switch (args.event) {	14,648✔
305	case 'progress':
306	args.data = JSON.parse(args.data);	510✔
307	break;	510✔
308	case 'completed':
309	args.returnvalue = JSON.parse(args.returnvalue);	1,848✔
310	break;	1,848✔
311	}
312
313	const { event, ...restArgs } = args;	14,648✔
314
315	if (event === 'drained') {	14,648✔
316	this.emit(event, id);	227✔
317	} else {
318	this.emit(event as any, restArgs, id);	14,421✔
319	if (restArgs.jobId) {	14,421✔
320	this.emit(`${event}:${restArgs.jobId}` as any, restArgs, id);	14,369✔
321	}
322	}
323	}
324	}
325	}
326	}
327
328	/**
329	* Stops consuming events and close the underlying Redis connection if necessary.
330	*
331	* @returns
332	*/
333	close(): Promise<void> {
334	if (!this.closing) {	376!
335	this.closing = this.disconnect();	376✔
336	}
337	return this.closing;	376✔
338	}
339	}

taskforcesh / bullmq / 11473539716

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