feat(amqp): add heartbeat and reconnect time configuration options (#2572)

This PR introduces configuration options for the AMQP heartbeat interval and reconnect time, improving connection stability and management. Previously, these values were not configurable, which could result in the client prematurely closing the connection.

Author: PooyaRaki

.env.sample

@@ -129,6 +129,12 @@
 # The message prefetch amount (max non-acknowledged messages being processed)
 # (default=100)
 # AMQP_PREFETCH=
+# The AMQP heartbeat interval in seconds
+# (default=60)
+# AMQP_HEARBEAT_INTERVAL_SECONDS=
+# The AMQP reconnect time in seconds
+# (default=5)
+# AMQP_RECONNECT_TIME_SECONDS=
 
 # Authorization token to use privileged endpoints.
 # The AUTH_TOKEN should always be set

src/__tests__/amqp-client.factory.ts

@@ -6,15 +6,27 @@ export function amqpClientFactory(queue?: string): {
   channel: ChannelWrapper;
   queueName: string;
 } {
-  const { AMQP_URL, AMQP_EXCHANGE_NAME, AMQP_EXCHANGE_MODE, AMQP_QUEUE } =
-    process.env;
+  const {
+    AMQP_URL,
+    AMQP_EXCHANGE_NAME,
+    AMQP_EXCHANGE_MODE,
+    AMQP_QUEUE,
+    AMQP_HEARBEAT_INTERVAL_SECONDS,
+    AMQP_RECONNECT_TIME_SECONDS,
+  } = process.env;
+
+  const heartbeatIntervalInSeconds = +(AMQP_HEARBEAT_INTERVAL_SECONDS || 60);
+  const reconnectTimeInSeconds = +(AMQP_RECONNECT_TIME_SECONDS || 5);
 
   if (!AMQP_URL || !AMQP_EXCHANGE_NAME || !AMQP_EXCHANGE_MODE || !AMQP_QUEUE) {
     throw new Error('Invalid amqpClientFactory configuration');
   }
 
   const queueName = queue ?? AMQP_QUEUE;
-  const connection = amqp.connect(AMQP_URL);
+  const connection = amqp.connect(AMQP_URL, {
+    heartbeatIntervalInSeconds,
+    reconnectTimeInSeconds,
+  });
   const channel = connection.createChannel({
     json: true,
     setup: async (ch: Channel) => {

src/config/entities/__tests__/configuration.ts

@@ -33,6 +33,8 @@ export default (): ReturnType<typeof configuration> => ({
     exchange: { name: faker.string.sample(), mode: faker.string.sample() },
     queue: faker.string.sample(),
     prefetch: faker.number.int(),
+    heartbeatIntervalInSeconds: 60,
+    reconnectTimeInSeconds: 5,
   },
   application: {
     isProduction: faker.datatype.boolean(),

src/config/entities/configuration.ts

@@ -59,6 +59,10 @@ export default () => ({
       process.env.AMQP_PREFETCH != null
         ? parseInt(process.env.AMQP_PREFETCH)
         : 100,
+    heartbeatIntervalInSeconds: +(
+      process.env.AMQP_HEARBEAT_INTERVAL_SECONDS || 60
+    ),
+    reconnectTimeInSeconds: +(process.env.AMQP_RECONNECT_TIME_SECONDS || 5),
   },
   application: {
     isProduction: process.env.CGW_ENV === 'production',

src/datasources/queues/queues-api.module.ts

@@ -25,8 +25,17 @@ function queueConsumerFactory(
     configurationService.getOrThrow<string>('amqp.exchange.mode');
   const queue = configurationService.getOrThrow<string>('amqp.queue');
   const prefetch = configurationService.getOrThrow<number>('amqp.prefetch');
+  const heartbeatIntervalInSeconds = configurationService.getOrThrow<number>(
+    'amqp.heartbeatIntervalInSeconds',
+  );
+  const reconnectTimeInSeconds = configurationService.getOrThrow<number>(
+    'amqp.reconnectTimeInSeconds',
+  );
 
-  const connection = amqp.connect(amqpUrl);
+  const connection = amqp.connect(amqpUrl, {
+    heartbeatIntervalInSeconds,
+    reconnectTimeInSeconds,
+  });
   const channel = connection.createChannel({
     json: true,
     setup: async (ch: Channel) => {