feat: add retry and back-off when creating receivers and senders

src/interface/queue-module-options.interface.ts

@@ -71,4 +71,17 @@ export interface AMQPConnectionOptions {
    * Connection options directly used by `rhea`
    */
   connectionOptions?: ConnectionOptions;
+
+  /**
+   * Retry configuration for senders and receivers
+   */
+  retryConnection?: {
+    receiver?: RetryConfig;
+    sender?: RetryConfig;
+  };
+}
+
+export interface RetryConfig {
+  retryDelay?: number;
+  maxRetryAttempts?: number;
 }

src/service/queue/queue.service.spec.ts

@@ -64,6 +64,19 @@ describe('QueueService', () => {
             createReceiver: jest.fn().mockResolvedValue(jest.fn().mockResolvedValue(new EventContextMock().receiver)),
             createSender: jest.fn().mockResolvedValue(new EventContextMock().sender),
             disconnect: jest.fn().mockResolvedValue(jest.fn()),
+            getConnectionOptions: jest.fn(() => ({
+              connectionUri: 'amqp://test',
+              retryConnection: {
+                receiver: {
+                  retryDelay: 1000,
+                  maxRetryAttempts: 3,
+                },
+                sender: {
+                  retryDelay: 1000,
+                  maxRetryAttempts: 3,
+                },
+              },
+            })),
             getModuleOptions(): QueueModuleOptions {
               return moduleOptions;
             },
@@ -241,6 +254,83 @@ describe('QueueService', () => {
         const messageControl = getInternallyCreatedMessageControl();
         expect(messageControl.accept).toHaveBeenCalled();
       });
+
+      it('should return an existing receiver if already created', async () => {
+        const receiver = {} as Receiver;
+        const source = 'test-queue';
+        queueService['receivers'].set('default:test-queue', receiver);
+
+        const result = await queueService['getReceiver'](source, 1, jest.fn(), 'default');
+
+        expect(result).toBe(receiver);
+        expect(amqpService.createReceiver).not.toHaveBeenCalled();
+      });
+
+      it('should create a new receiver if not already created', async () => {
+        const receiver = {} as Receiver;
+        const source = 'test-queue';
+        const messageHandler = jest.fn();
+
+        (amqpService as any).createReceiver.mockResolvedValue(receiver);
+
+        const result = await queueService['getReceiver'](source, 1, messageHandler, 'default');
+
+        expect(result).toBe(receiver);
+        expect(amqpService.createReceiver).toHaveBeenCalledWith(source, 1, expect.any(Function), 'default');
+      });
+
+      it('should retry creating a receiver on failure', async () => {
+        const source = 'test-queue';
+        const messageHandler = jest.fn();
+
+        (amqpService as any).createReceiver.mockRejectedValueOnce(new Error('Test error')).mockResolvedValueOnce({} as Receiver);
+
+        const result = await queueService['getReceiver'](source, 1, messageHandler, 'default');
+
+        expect(result).toBeDefined();
+        expect(amqpService.createReceiver).toHaveBeenCalledTimes(2);
+      });
+
+      it('should not retry creating a receiver if maxRetryAttempts is 1', async () => {
+        (amqpService as any).getConnectionOptions.mockReturnValueOnce({
+          retryConnection: {
+            receiver: {
+              retryDelay: 1000,
+              maxRetryAttempts: 1,
+            },
+          },
+        });
+
+        const source = 'test-queue';
+        const messageHandler = jest.fn();
+
+        (amqpService as any).createReceiver.mockRejectedValue(new Error('Test error'));
+
+        await expect(queueService['getReceiver'](source, 1, messageHandler, 'default')).rejects.toThrow('Test error');
+
+        expect(amqpService.createReceiver).toHaveBeenCalledTimes(1);
+      });
+
+      it('should not retry if both retryDelay and maxRetryAttempts are set to zero', async () => {
+        (amqpService.getConnectionOptions as jest.Mock).mockReturnValueOnce({
+          connectionUri: 'amqp://test',
+          retryConnection: {
+            receiver: {
+              retryDelay: 0,
+              maxRetryAttempts: 0,
+            },
+          },
+        });
+
+        const source = 'test-queue';
+        const messageHandler = jest.fn();
+
+        (amqpService.createReceiver as jest.Mock).mockRejectedValue(new Error('Test error'));
+
+        await expect(queueService['getReceiver'](source, 1, messageHandler, 'default')).rejects.toThrow('Test error');
+
+        expect(amqpService.createReceiver).toHaveBeenCalledTimes(1);
+      });
     });
   });
 
@@ -347,6 +437,81 @@ describe('QueueService', () => {
 
       expect(sender.send).toHaveBeenCalledWith({ body: 'null', message_annotations: { 'x-opt-delivery-delay': delay * 1000 } });
     });
+
+    it('should return an existing sender if already created', async () => {
+      const sender = {} as AwaitableSender;
+      const target = 'test-queue';
+      queueService['senders'].set('default:test-queue', sender);
+
+      const result = await queueService['getSender'](target, 'default');
+
+      expect(result).toBe(sender);
+      expect(amqpService.createSender).not.toHaveBeenCalled();
+    });
+
+    it('should create a new sender if not already created', async () => {
+      const sender = {} as AwaitableSender;
+      const target = 'test-queue';
+
+      (amqpService as any).createSender.mockResolvedValue(sender);
+
+      const result = await queueService['getSender'](target, 'default');
+
+      expect(result).toBe(sender);
+      expect(amqpService.createSender).toHaveBeenCalledWith(target, 'default');
+    });
+
+    it('should retry creating a sender on failure', async () => {
+      const target = 'test-queue';
+
+      (amqpService as any).createSender.mockRejectedValueOnce(new Error('Test error')).mockResolvedValueOnce({} as AwaitableSender);
+
+      const result = await queueService['getSender'](target, 'default');
+
+      expect(result).toBeDefined();
+      expect(amqpService.createSender).toHaveBeenCalledTimes(2);
+    });
+
+    it('should retry creating a sender with custom retry configuration', async () => {
+      (amqpService.getConnectionOptions as jest.Mock).mockReturnValueOnce({
+        connectionUri: 'amqp://test',
+        retryConnection: {
+          sender: {
+            retryDelay: 500,
+            maxRetryAttempts: 2,
+          },
+        },
+      });
+
+      const target = 'test-queue';
+
+      (amqpService.createSender as jest.Mock).mockRejectedValueOnce(new Error('Test error')).mockResolvedValueOnce({} as AwaitableSender);
+
+      const result = await queueService['getSender'](target, 'default');
+
+      expect(result).toBeDefined();
+      expect(amqpService.createSender).toHaveBeenCalledTimes(2);
+    });
+
+    it('should not retry creating a sender if maxRetryAttempts is 1', async () => {
+      (amqpService.getConnectionOptions as jest.Mock).mockReturnValueOnce({
+        connectionUri: 'amqp://test',
+        retryConnection: {
+          sender: {
+            retryDelay: 1000,
+            maxRetryAttempts: 1,
+          },
+        },
+      });
+
+      const target = 'test-queue';
+
+      (amqpService.createSender as jest.Mock).mockRejectedValue(new Error('Test error'));
+
+      await expect(queueService['getSender'](target, 'default')).rejects.toThrow('Test error');
+
+      expect(amqpService.createSender).toHaveBeenCalledTimes(1);
+    });
   });
 
   describe('removeListener()', () => {

src/service/queue/queue.service.ts

@@ -4,12 +4,12 @@ import { AwaitableSender, Delivery, EventContext, Message, Receiver, Source } fr
 
 import {
   extendObject,
+  getLoggerContext,
+  Logger,
   sleep,
   tryParseJSON,
-  ValidationNullObjectException,
-  Logger,
-  getLoggerContext,
   ValidationException,
+  ValidationNullObjectException,
 } from '../../util';
 import { MessageControl } from '../../domain';
 import { SendState } from '../../enum';
@@ -331,31 +331,69 @@ export class QueueService {
     connection: string,
   ): Promise<Receiver> {
     const sourceToken = typeof source === 'string' ? source : JSON.stringify(source);
-
     const receiverToken = this.getLinkToken(sourceToken, connection);
 
-    if (!this.receivers.has(receiverToken)) {
-      const receiver = await this.amqpService.createReceiver(source, credit, messageHandler.bind(this), connection);
-      this.receivers.set(receiverToken, receiver);
+    if (this.receivers.has(receiverToken)) {
+      return this.receivers.get(receiverToken);
     }
 
-    return this.receivers.get(receiverToken);
+    const connectionOptions = this.amqpService.getConnectionOptions(connection);
+    const retryDelay = connectionOptions.retryConnection?.receiver?.retryDelay ?? 0;
+    const maxRetryAttempts = connectionOptions.retryConnection?.receiver?.maxRetryAttempts ?? 1;
+
+    let attempt = 0;
+
+    do {
+      try {
+        const receiver = await this.amqpService.createReceiver(source, credit, messageHandler.bind(this), connection);
+        this.receivers.set(receiverToken, receiver);
+        return receiver;
+      } catch (error) {
+        logger.error(`Error creating receiver (attempt ${attempt + 1}): ${error.message}`, error.stack);
+
+        attempt = attempt + 1;
+        if (attempt >= maxRetryAttempts) {
+          throw new Error(`Max retry attempts reached for creating receiver: ${error.message}`);
+        }
+
+        if (retryDelay > 0) {
+          await sleep(retryDelay);
+        }
+      }
+    } while (attempt < maxRetryAttempts);
   }
 
   private async getSender(target: string, connection: string): Promise<AwaitableSender> {
-    let sender;
-
     const senderToken = this.getLinkToken(target, connection);
 
     if (this.senders.has(senderToken)) {
-      sender = this.senders.get(senderToken);
-    } else {
-      sender = await this.amqpService.createSender(target, connection);
-
-      this.senders.set(senderToken, sender);
+      return this.senders.get(senderToken);
     }
 
-    return sender;
+    const connectionOptions = this.amqpService.getConnectionOptions(connection);
+    const retryDelay = connectionOptions.retryConnection?.sender?.retryDelay ?? 0;
+    const maxRetryAttempts = connectionOptions.retryConnection?.sender?.maxRetryAttempts ?? 1;
+
+    let attempt = 0;
+
+    do {
+      try {
+        const sender = await this.amqpService.createSender(target, connection);
+        this.senders.set(senderToken, sender);
+        return sender;
+      } catch (error) {
+        logger.error(`Error creating sender (attempt ${attempt + 1}): ${error.message}`, error.stack);
+
+        attempt++;
+        if (attempt >= maxRetryAttempts) {
+          throw new Error(`Max retry attempts reached for creating sender: ${error.message}`);
+        }
+
+        if (retryDelay > 0) {
+          await sleep(retryDelay);
+        }
+      }
+    } while (attempt < maxRetryAttempts);
   }
 
   private encodeMessage(message: any): string {