Re-queue with with x-delay

[aviNrich]

Re-queue with with x-delay

[brknesn]

aviNrich I apply it this way.

I created a decorator named Requeueable:

export function Requeueable(maxRetries = 3, delayInMillis = 5000) {
    return function (target: any, propertyKey: string, descriptor: PropertyDescriptor) {
        const originalMethod = descriptor.value;

        descriptor.value = async function (...args: any[]) {
            try {
                await originalMethod.apply(this, args);
            } catch (error) {
                const message = args[1]; // The message is the second argument to the method
                const retryCount = message.properties.headers?.retryCount || 0;

                await sleep(delayInMillis * retryCount);

                if (retryCount < maxRetries) {
                    console.error(`Error: ${error.message}, requeueing message. Retry Count: ${retryCount + 1}`);

                    const amqpConnection: AmqpConnection = this.amqpConnection; // Ensure AmqpConnection is injected into your service
                    await amqpConnection.publish(
                        message.fields.exchange,
                        message.fields.routingKey,
                        args[0],
                        {
                            headers: {
                                ...message.properties.headers,
                                retryCount: retryCount + 1,
                                'x-delay': delayInMillis * retryCount,
                            }
                        }
                    );

                    return new Nack();
                } else {
                    console.error(`Maximum retry count reached. Message will be discarded.`);
                    return new Nack();
                }
            }
        };

        return descriptor;
    };
}

And in my service, I use it like this:

const MAX_RETRIES = 50;
const RETRY_DELAY = 2000;

@Injectable()
export class RabbitmqHandler {
  constructor(
    private readonly amqpConnection: AmqpConnection,
  ) {}

  @RabbitSubscribe({
    exchange: 'initiation_exchange',
    routingKey: 'ScanReportCreatedEvent',
    queue: 'product-creation-queue',
    queueOptions: {
      durable: true,
      channel: 'product-creation-channel',
    },
    allowNonJsonMessages: true,
    createQueueIfNotExists: true,
  })
  @Requeueable(MAX_RETRIES, RETRY_DELAY)
  public async pubSubHandler(msg: {}, amqpMsg: ConsumeMessage) {
    throw new Error('test');
    }

[underfisk]

@brknesn Thank you for providing your Requeueable decorator