8649369533

Committed 11 Apr 2024 03:32PM UTC coverage: 92.904% (-0.1%) from 93.001%

Build # 8649369533

Build Type

Pull #1382

github

Committed by

hectorgomezv

Commit Message

Refactor, add documentation

Pull Request Pull Request #1382: [WIP] PoC: consume events from TXS queue

Run Details

1941 of 2347 branches covered (82.7%)

Branch coverage included in aggregate %.

90 of 96 new or added lines in 12 files covered. (93.75%)

9 existing lines in 2 files now uncovered.

6779 of 7039 relevant lines covered (96.31%)

359.15 hits per line

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

75.0

/src/datasources/queues/queue-consumer.service.ts

import { IConfigurationService } from '@/config/configuration.service.interface';
import { QueueConsumer } from '@/datasources/queues/queue-consumer.module';
import { IQueueConsumerService } from '@/datasources/queues/queue-consumer.service.interface';
import { ILoggingService, LoggingService } from '@/logging/logging.interface';
import { Inject } from '@nestjs/common';
import { ConsumeMessage } from 'amqplib';

export class QueueConsumerService implements IQueueConsumerService {
  private readonly isEventsQueueEnabled: boolean;

  constructor(
    @Inject('QueueConsumer') private readonly consumer: QueueConsumer,
    @Inject(LoggingService) private readonly loggingService: ILoggingService,
    @Inject(IConfigurationService)
    private readonly configurationService: IConfigurationService,
  ) {
    this.isEventsQueueEnabled = this.configurationService.getOrThrow<boolean>(
      'features.eventsQueue',
    );
  }

  isReady(): boolean {
    if (this.isEventsQueueEnabled) {
      return this.consumer.connection.isConnected();
    }
    return true;
  }

  async subscribe(
    queueName: string,
    fn: (msg: ConsumeMessage) => Promise<void>,
  ): Promise<void> {
    if (this.isEventsQueueEnabled) {
      await this.consumer.channel.consume(queueName, async (msg) => {
        await fn(msg);
        await this.consumer.channel.ack(msg);
      });
      this.loggingService.info(`Subscribed to queue: ${queueName}`);
    }
  }
}

1	import { IConfigurationService } from '@/config/configuration.service.interface';	94✔
2	import { QueueConsumer } from '@/datasources/queues/queue-consumer.module';
3	import { IQueueConsumerService } from '@/datasources/queues/queue-consumer.service.interface';
4	import { ILoggingService, LoggingService } from '@/logging/logging.interface';	94✔
5	import { Inject } from '@nestjs/common';	94✔
6	import { ConsumeMessage } from 'amqplib';
7
8	export class QueueConsumerService implements IQueueConsumerService {	94✔
9	private readonly isEventsQueueEnabled: boolean;
10
11	constructor(
12	@Inject('QueueConsumer') private readonly consumer: QueueConsumer,	28✔
13	@Inject(LoggingService) private readonly loggingService: ILoggingService,	28✔
14	@Inject(IConfigurationService)
15	private readonly configurationService: IConfigurationService,	28✔
16	) {
17	this.isEventsQueueEnabled = this.configurationService.getOrThrow<boolean>(	28✔
18	'features.eventsQueue',
19	);
20	}
21
22	isReady(): boolean {
NEW 23	if (this.isEventsQueueEnabled) {	×
NEW 24	return this.consumer.connection.isConnected();	×
25	}
NEW 26	return true;	×
27	}
28
29	async subscribe(
30	queueName: string,
31	fn: (msg: ConsumeMessage) => Promise<void>,
32	): Promise<void> {
33	if (this.isEventsQueueEnabled) {	28✔
34	await this.consumer.channel.consume(queueName, async (msg) => {	2✔
35	await fn(msg);	66✔
36	await this.consumer.channel.ack(msg);	66✔
37	});
38	this.loggingService.info(`Subscribed to queue: ${queueName}`);	2✔
39	}
40	}
41	}

safe-global / safe-client-gateway / 8649369533

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