8631224700

Committed 10 Apr 2024 12:28PM UTC coverage: 92.332% (-0.3%) from 92.633%

Build # 8631224700

Build Type

Pull #1382

github

Committed by

hectorgomezv

Commit Message

Fix DI in HealthController tests

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

Run Details

1895 of 2295 branches covered (82.57%)

Branch coverage included in aggregate %.

46 of 77 new or added lines in 10 files covered. (59.74%)

2 existing lines in 2 files now uncovered.

6642 of 6951 relevant lines covered (95.55%)

354.1 hits per line

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

50.0

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

import { IConfigurationService } from '@/config/configuration.service.interface';
import configuration from '@/config/entities/configuration';
import { QueueConsumerService } from '@/datasources/queues/queue-consumer.service';
import { IQueueConsumerService } from '@/datasources/queues/queue-consumer.service.interface';
import { QueueConsumerShutdownHook } from '@/datasources/queues/queue-consumer.shutdown.hook';
import { DynamicModule, Module } from '@nestjs/common';
import amqp, {
  AmqpConnectionManager,
  ChannelWrapper,
} from 'amqp-connection-manager';
import { Channel } from 'amqplib';

export interface QueueConsumer {
  connection: AmqpConnectionManager;
  channel: ChannelWrapper;
}

function queueConsumerFactory(
  configurationService: IConfigurationService,
): QueueConsumer {
  const amqpUrl = configurationService.getOrThrow<string>('amqp.url');
  const exchangeName =
    configurationService.getOrThrow<string>('amqp.exchange.name');
  const exchangeMode =
    configurationService.getOrThrow<string>('amqp.exchange.mode');
  const queue = configurationService.getOrThrow<string>('amqp.queue');
  const prefetch = configurationService.getOrThrow<number>('amqp.prefetch');

  const connection = amqp.connect(amqpUrl);
  const channel = connection.createChannel({
    json: true,
    setup: async (ch: Channel) => {
      await ch.assertExchange(exchangeName, exchangeMode, { durable: true });
      await ch.assertQueue(queue, { durable: true });
      await ch.prefetch(prefetch);
      await ch.bindQueue(queue, exchangeName, '');
    },
  });
  return { connection, channel };
}

@Module({})
export class QueueConsumerModule {
  static register(configFactory = configuration): DynamicModule {
    const { eventsQueue: isEventsQueueEnabled } = configFactory()['features'];
    return {
      module: QueueConsumerModule,
      providers: isEventsQueueEnabled
        ? [
            {
              provide: 'QueueConsumer',
              useFactory: queueConsumerFactory,
              inject: [IConfigurationService],
            },
            { provide: IQueueConsumerService, useClass: QueueConsumerService },
            QueueConsumerShutdownHook,
          ]
        : [
            {
              provide: IQueueConsumerService,
              useFactory: () => ({
                isReady: () => true,
                subscribe: () => Promise.resolve(),
              }),
            },
          ],
      exports: [IQueueConsumerService],
    };
  }
}

1	import { IConfigurationService } from '@/config/configuration.service.interface';	90✔
2	import configuration from '@/config/entities/configuration';	90✔
3	import { QueueConsumerService } from '@/datasources/queues/queue-consumer.service';	90✔
4	import { IQueueConsumerService } from '@/datasources/queues/queue-consumer.service.interface';	90✔
5	import { QueueConsumerShutdownHook } from '@/datasources/queues/queue-consumer.shutdown.hook';	90✔
6	import { DynamicModule, Module } from '@nestjs/common';	90✔
7	import amqp, {	90✔
8	AmqpConnectionManager,
9	ChannelWrapper,
10	} from 'amqp-connection-manager';
11	import { Channel } from 'amqplib';
12
13	export interface QueueConsumer {
14	connection: AmqpConnectionManager;
15	channel: ChannelWrapper;
16	}
17
18	function queueConsumerFactory(
19	configurationService: IConfigurationService,
20	): QueueConsumer {
NEW 21	const amqpUrl = configurationService.getOrThrow<string>('amqp.url');	×
22	const exchangeName =
NEW 23	configurationService.getOrThrow<string>('amqp.exchange.name');	×
24	const exchangeMode =
NEW 25	configurationService.getOrThrow<string>('amqp.exchange.mode');	×
NEW 26	const queue = configurationService.getOrThrow<string>('amqp.queue');	×
NEW 27	const prefetch = configurationService.getOrThrow<number>('amqp.prefetch');	×
28
NEW 29	const connection = amqp.connect(amqpUrl);	×
NEW 30	const channel = connection.createChannel({	×
31	json: true,
32	setup: async (ch: Channel) => {
NEW 33	await ch.assertExchange(exchangeName, exchangeMode, { durable: true });	×
NEW 34	await ch.assertQueue(queue, { durable: true });	×
NEW 35	await ch.prefetch(prefetch);	×
NEW 36	await ch.bindQueue(queue, exchangeName, '');	×
37	},
38	});
NEW 39	return { connection, channel };	×
40	}
41
42	@Module({})
43	export class QueueConsumerModule {	90✔
44	static register(configFactory = configuration): DynamicModule {	90✔
45	const { eventsQueue: isEventsQueueEnabled } = configFactory()['features'];	180✔
46	return {	180✔
47	module: QueueConsumerModule,
48	providers: isEventsQueueEnabled	90!
49	? [
50	{
51	provide: 'QueueConsumer',
52	useFactory: queueConsumerFactory,
53	inject: [IConfigurationService],
54	},
55	{ provide: IQueueConsumerService, useClass: QueueConsumerService },
56	QueueConsumerShutdownHook,
57	]
58	: [
59	{
60	provide: IQueueConsumerService,
61	useFactory: () => ({	1,188✔
NEW 62	isReady: () => true,	×
63	subscribe: () => Promise.resolve(),	1,188✔
64	}),
65	},
66	],
67	exports: [IQueueConsumerService],
68	};
69	}
70	}

safe-global / safe-client-gateway / 8631224700

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