9992950173

Committed 18 Jul 2024 02:01PM UTC coverage: 48.271% (+0.03%) from 48.244%

Build # 9992950173

Build Type

Pull #1774

github

Committed by

iamacook

Commit Message

Merge branch 'main' into multiple-queue-subscriptions

Pull Request Pull Request #1774: Accept multiple listeners for the same queue subscription

Run Details

417 of 2593 branches covered (16.08%)

Branch coverage included in aggregate %.

17 of 18 new or added lines in 3 files covered. (94.44%)

3 existing lines in 2 files now uncovered.

4286 of 7150 relevant lines covered (59.94%)

13.0 hits per line

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

94.74

/src/domain/queues/queues-repository.ts

import { IConfigurationService } from '@/config/configuration.service.interface';
import { IQueuesApiService } from '@/datasources/queues/queues-api.service.interface';
import { IQueuesRepository } from '@/domain/queues/queues-repository.interface';
import { ILoggingService, LoggingService } from '@/logging/logging.interface';
import { WebHookSchema } from '@/routes/cache-hooks/entities/schemas/web-hook.schema';
import { Event } from '@/routes/cache-hooks/entities/event.entity';
import { Inject, Injectable, Scope } from '@nestjs/common';
import { ConsumeMessage } from 'amqplib';

// Only allow one instance across the app to prevent multiple queue subscriptions
@Injectable({ scope: Scope.DEFAULT })
export class QueuesRepository implements IQueuesRepository {
  private readonly queueName: string;
  private readonly listeners: Array<(event: Event) => Promise<unknown>> = [];

  constructor(
    @Inject(IQueuesApiService) private readonly queuesApi: IQueuesApiService,
    @Inject(IConfigurationService)
    private readonly configurationService: IConfigurationService,
    @Inject(LoggingService) private readonly loggingService: ILoggingService,
  ) {
    this.queueName = this.configurationService.getOrThrow<string>('amqp.queue');
  }

  onModuleInit(): Promise<void> {
    return this.queuesApi.subscribe(
      this.queueName,
      async (msg: ConsumeMessage): Promise<void> => {
        try {
          const content = JSON.parse(msg.content.toString());
          const event = WebHookSchema.parse(content);
          await Promise.allSettled(
            this.listeners.map((listener) => listener(event)),
          );
        } catch (err) {
          this.loggingService.error(err);
        }
      },
    );
  }

  onEvent(listener: (event: Event) => Promise<unknown>): void {
    this.listeners.push(listener);
  }
}

1	import { IConfigurationService } from '@/config/configuration.service.interface';	16✔
2	import { IQueuesApiService } from '@/datasources/queues/queues-api.service.interface';	16✔
3	import { IQueuesRepository } from '@/domain/queues/queues-repository.interface';
4	import { ILoggingService, LoggingService } from '@/logging/logging.interface';	16✔
5	import { WebHookSchema } from '@/routes/cache-hooks/entities/schemas/web-hook.schema';	16✔
6	import { Event } from '@/routes/cache-hooks/entities/event.entity';
7	import { Inject, Injectable, Scope } from '@nestjs/common';	16✔
8	import { ConsumeMessage } from 'amqplib';
9
10	// Only allow one instance across the app to prevent multiple queue subscriptions
11	@Injectable({ scope: Scope.DEFAULT })
12	export class QueuesRepository implements IQueuesRepository {	16✔
13	private readonly queueName: string;
14	private readonly listeners: Array<(event: Event) => Promise<unknown>> = [];	16✔
15
16	constructor(
17	@Inject(IQueuesApiService) private readonly queuesApi: IQueuesApiService,	16✔
18	@Inject(IConfigurationService)
19	private readonly configurationService: IConfigurationService,	16✔
20	@Inject(LoggingService) private readonly loggingService: ILoggingService,	16✔
21	) {
22	this.queueName = this.configurationService.getOrThrow<string>('amqp.queue');	16✔
23	}
24
25	onModuleInit(): Promise<void> {
26	return this.queuesApi.subscribe(	16✔
27	this.queueName,
28	async (msg: ConsumeMessage): Promise<void> => {
29	try {	66✔
30	const content = JSON.parse(msg.content.toString());	66✔
31	const event = WebHookSchema.parse(content);	66✔
32	await Promise.allSettled(	66✔
33	this.listeners.map((listener) => listener(event)),	66✔
34	);
35	} catch (err) {
NEW UNCOV 36	this.loggingService.error(err);	×
37	}
38	},
39	);
40	}
41
42	onEvent(listener: (event: Event) => Promise<unknown>): void {
43	this.listeners.push(listener);	16✔
44	}
45	}

safe-global / safe-client-gateway / 9992950173

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