Adding HTTP Client and TTLPolicyManager with fallback to queue-ttl

src/core/RunMQ.ts

@@ -2,7 +2,7 @@ import {RunMQProcessorConfiguration, RunMQConnectionConfig, RunMQPublisher, RunM
 import {RunMQException} from "@src/core/exceptions/RunMQException";
 import {AmqplibClient} from "@src/core/clients/AmqplibClient";
 import {Exceptions} from "@src/core/exceptions/Exceptions";
-import {RunMQUtils} from "@src/core/utils/Utils";
+import {RunMQUtils} from "@src/core/utils/RunMQUtils";
 import {Constants, DEFAULTS} from "@src/core/constants";
 import {Channel} from "amqplib";
 import {RunMQConsumerCreator} from "@src/core/consumer/RunMQConsumerCreator";
@@ -51,7 +51,7 @@ export class RunMQ {
      * @param processor The function that will process the incoming messages
      */
     public async process<T = Record<string, never>>(topic: string, config: RunMQProcessorConfiguration, processor: (message: RunMQMessageContent<T>) => Promise<void>) {
-        const consumer = new RunMQConsumerCreator(this.defaultChannel!, this.amqplibClient, this.logger);
+        const consumer = new RunMQConsumerCreator(this.defaultChannel!, this.amqplibClient, this.logger, this.config.management);
         await consumer.createConsumer<T>(new ConsumerConfiguration(topic, config, processor))
     }
 

src/core/constants/index.ts

@@ -4,6 +4,7 @@ export const Constants = {
     DEAD_LETTER_ROUTER_EXCHANGE_NAME: RUNMQ_PREFIX + "dead_letter_router",
     RETRY_DELAY_QUEUE_PREFIX: RUNMQ_PREFIX + "retry_delay_",
     DLQ_QUEUE_PREFIX: RUNMQ_PREFIX + "dlq_",
+    MESSAGE_TTL_OPERATOR_POLICY_PREFIX: RUNMQ_PREFIX + "message_ttl_operator_policy",
 }
 
 export const DEFAULTS = {

src/core/consumer/ConsumerCreatorUtils.ts

@@ -7,4 +7,8 @@ export class ConsumerCreatorUtils {
     static getRetryDelayTopicName(topic: string): string {
         return Constants.RETRY_DELAY_QUEUE_PREFIX + topic;
     }
+
+    static getMessageTTLPolicyName(topic: string): string {
+        return Constants.MESSAGE_TTL_OPERATOR_POLICY_PREFIX + topic;
+    }
 }

src/core/consumer/RunMQConsumerCreator.ts

@@ -14,18 +14,25 @@ import {RunMQLogger} from "@src/core/logging/RunMQLogger";
 import {DefaultDeserializer} from "@src/core/serializers/deserializer/DefaultDeserializer";
 import {ConsumerCreatorUtils} from "@src/core/consumer/ConsumerCreatorUtils";
 import {RunMQPublisherCreator} from "@src/core/publisher/RunMQPublisherCreator";
-import {AMQPClient} from "@src/types";
+import {AMQPClient, RabbitMQManagementConfig} from "@src/types";
+import {RunMQTTLPolicyManager} from "@src/core/management/Policies/RunMQTTLPolicyManager";
+import {RunMQException} from "@src/core/exceptions/RunMQException";
+import {Exceptions} from "@src/core/exceptions/Exceptions";
 
 export class RunMQConsumerCreator {
+    private ttlPolicyManager: RunMQTTLPolicyManager;
+
     constructor(
         private defaultChannel: Channel,
         private client: AMQPClient,
         private logger: RunMQLogger,
+        managementConfig?: RabbitMQManagementConfig
     ) {
+        this.ttlPolicyManager = new RunMQTTLPolicyManager(logger, managementConfig);
     }
 
-
     public async createConsumer<T>(consumerConfiguration: ConsumerConfiguration<T>) {
+        await this.ttlPolicyManager.initialize();
         await this.assertQueues<T>(consumerConfiguration);
         await this.bindQueues<T>(consumerConfiguration);
         for (let i = 0; i < consumerConfiguration.processorConfig.consumersCount; i++) {
@@ -78,16 +85,43 @@ export class RunMQConsumerCreator {
             deadLetterExchange: Constants.DEAD_LETTER_ROUTER_EXCHANGE_NAME,
             deadLetterRoutingKey: consumerConfiguration.processorConfig.name
         });
-        await this.defaultChannel.assertQueue(ConsumerCreatorUtils.getRetryDelayTopicName(consumerConfiguration.processorConfig.name), {
-            durable: true,
-            deadLetterExchange: Constants.ROUTER_EXCHANGE_NAME,
-            messageTtl: consumerConfiguration.processorConfig.attemptsDelay ?? DEFAULTS.PROCESSING_RETRY_DELAY,
-        });
         await this.defaultChannel.assertQueue(ConsumerCreatorUtils.getDLQTopicName(consumerConfiguration.processorConfig.name), {
             durable: true,
             deadLetterExchange: Constants.ROUTER_EXCHANGE_NAME,
             deadLetterRoutingKey: consumerConfiguration.processorConfig.name
         });
+
+        const retryDelayQueueName = ConsumerCreatorUtils.getRetryDelayTopicName(consumerConfiguration.processorConfig.name);
+        const messageDelay = consumerConfiguration.processorConfig.attemptsDelay ?? DEFAULTS.PROCESSING_RETRY_DELAY
+
+
+        const policiesForTTL = consumerConfiguration.processorConfig.usePoliciesForDelay ?? false;
+        if (!policiesForTTL) {
+            await this.defaultChannel.assertQueue(retryDelayQueueName, {
+                durable: true,
+                deadLetterExchange: Constants.ROUTER_EXCHANGE_NAME,
+                messageTtl: messageDelay,
+            });
+            return;
+        }
+
+        const result = await this.ttlPolicyManager.apply(
+            retryDelayQueueName,
+            messageDelay
+        );
+        if (result) {
+            await this.defaultChannel.assertQueue(retryDelayQueueName, {
+                durable: true,
+                deadLetterExchange: Constants.ROUTER_EXCHANGE_NAME
+            });
+            return;
+        }
+        throw new RunMQException(
+            Exceptions.FAILURE_TO_DEFINE_TTL_POLICY,
+            {
+                error: "Failed to apply TTL policy to queue: " + retryDelayQueueName
+            }
+        );
     }
 
 

src/core/exceptions/Exceptions.ts

@@ -4,4 +4,5 @@ export class Exceptions {
     public static NOT_INITIALIZED = 'NOT_INITIALIZED';
     public static INVALID_MESSAGE_FORMAT = 'MESSAGE_SHOULD_BE_VALID_RECORD';
     public static UNSUPPORTED_SCHEMA = 'UNSUPPORTED_SCHEMA';
+    public static FAILURE_TO_DEFINE_TTL_POLICY = 'FAILURE_TO_DEFINE_TTL_POLICY';
 }

src/core/management/Policies/RabbitMQMessageTTLPolicy.ts

@@ -0,0 +1,17 @@
+import {RabbitMQOperatorPolicy} from "@src/types";
+import {RunMQUtils} from "@src/core/utils/RunMQUtils";
+import {ConsumerCreatorUtils} from "@src/core/consumer/ConsumerCreatorUtils";
+
+export class RabbitMQMessageTTLPolicy {
+    static createFor(queueName: string, ttl: number): RabbitMQOperatorPolicy {
+        return {
+            name: ConsumerCreatorUtils.getMessageTTLPolicyName(queueName),
+            pattern: RunMQUtils.escapeRegExp(queueName),
+            definition: {
+                "message-ttl": ttl
+            },
+            "apply-to": "queues",
+            priority: 1000
+        }
+    }
+}

src/core/management/Policies/RunMQTTLPolicyManager.ts

@@ -0,0 +1,62 @@
+import {RabbitMQManagementClient} from "@src/core/management/RabbitMQManagementClient";
+import {RunMQLogger} from "@src/core/logging/RunMQLogger";
+import {DEFAULTS} from "@src/core/constants";
+import {RabbitMQManagementConfig} from "@src";
+import {RabbitMQMessageTTLPolicy} from "@src/core/management/Policies/RabbitMQMessageTTLPolicy";
+
+export class RunMQTTLPolicyManager {
+    private readonly managementClient: RabbitMQManagementClient | null = null;
+    private isManagementPluginEnabled = false;
+
+    constructor(
+        private logger: RunMQLogger,
+        private managementConfig?: RabbitMQManagementConfig
+    ) {
+        if (this.managementConfig) {
+            this.managementClient = new RabbitMQManagementClient(this.managementConfig, this.logger);
+        }
+    }
+
+    public async initialize(): Promise<void> {
+        if (!this.managementClient) {
+            this.logger.warn("Management client not configured");
+            return;
+        }
+
+        this.isManagementPluginEnabled = await this.managementClient.checkManagementPluginEnabled();
+
+        if (!this.isManagementPluginEnabled) {
+            this.logger.warn("RabbitMQ management plugin is not enabled");
+        } else {
+            this.logger.info("RabbitMQ management plugin is enabled");
+        }
+    }
+
+    public async apply(
+        queueName: string,
+        ttl?: number,
+        vhost: string = "%2F"
+    ): Promise<boolean> {
+        const actualTTL = ttl ?? DEFAULTS.PROCESSING_RETRY_DELAY;
+
+        if (this.isManagementPluginEnabled && this.managementClient) {
+            const success = await this.managementClient.createOrUpdateOperatorPolicy(
+                vhost,
+                RabbitMQMessageTTLPolicy.createFor(queueName, actualTTL)
+            );
+
+            if (success) {
+                return true
+            }
+        }
+        return false;
+    }
+
+
+    public async cleanup(queueName: string, vhost: string = "%2F"): Promise<void> {
+        if (this.isManagementPluginEnabled && this.managementClient) {
+            const policyName = `ttl-policy-${queueName}`;
+            await this.managementClient.deleteOperatorPolicy(vhost, policyName);
+        }
+    }
+}

src/core/management/RabbitMQManagementClient.ts

@@ -0,0 +1,117 @@
+import {RunMQLogger} from "@src/core/logging/RunMQLogger";
+import {RabbitMQManagementConfig} from "@src";
+import {RabbitMQOperatorPolicy} from "@src/types";
+
+export class RabbitMQManagementClient {
+    constructor(
+        private config: RabbitMQManagementConfig,
+        private logger: RunMQLogger
+    ) {}
+
+    private getAuthHeader(): string {
+        const credentials = Buffer.from(`${this.config.username}:${this.config.password}`).toString('base64');
+        return `Basic ${credentials}`;
+    }
+
+    public async createOrUpdateOperatorPolicy(vhost: string, policy: RabbitMQOperatorPolicy): Promise<boolean> {
+        try {
+            const url = `${this.config.url}/api/operator-policies/${vhost}/${encodeURIComponent(policy.name)}`;
+
+            const response = await fetch(url, {
+                method: 'PUT',
+                headers: {
+                    'Content-Type': 'application/json',
+                    'Authorization': this.getAuthHeader()
+                },
+                body: JSON.stringify({
+                    pattern: policy.pattern,
+                    definition: policy.definition,
+                    priority: policy.priority || 0,
+                    "apply-to": policy["apply-to"]
+                })
+            });
+
+            if (!response.ok) {
+                const error = await response.text();
+                this.logger.error(`Failed to create operator policy: ${response.status} - ${error}`);
+                return false;
+            }
+
+            this.logger.info(`Successfully set operator policy: ${policy.name}`);
+            return true;
+        } catch (error) {
+            this.logger.error(`Error creating operator policy: ${error}`);
+            return false;
+        }
+    }
+
+    public async getOperatorPolicy(vhost: string, policyName: string): Promise<RabbitMQOperatorPolicy | null> {
+        try {
+            const url = `${this.config.url}/api/operator-policies/${vhost}/${encodeURIComponent(policyName)}`;
+
+            const response = await fetch(url, {
+                method: 'GET',
+                headers: {
+                    'Authorization': this.getAuthHeader()
+                }
+            });
+
+            if (!response.ok) {
+                if (response.status === 404) {
+                    return null;
+                }
+                const error = await response.text();
+                this.logger.error(`Failed to get operator policy: ${response.status} - ${error}`);
+                return null;
+            }
+
+            return await response.json();
+        } catch (error) {
+            this.logger.error(`Error getting operator policy: ${error}`);
+            return null;
+        }
+    }
+
+    public async deleteOperatorPolicy(vhost: string, policyName: string): Promise<boolean> {
+        try {
+            const url = `${this.config.url}/api/operator-policies/${vhost}/${encodeURIComponent(policyName)}`;
+
+            const response = await fetch(url, {
+                method: 'DELETE',
+                headers: {
+                    'Authorization': this.getAuthHeader()
+                }
+            });
+
+            if (!response.ok && response.status !== 404) {
+                const error = await response.text();
+                this.logger.error(`Failed to delete operator policy: ${response.status} - ${error}`);
+                return false;
+            }
+
+            this.logger.info(`Successfully deleted operator policy: ${policyName}`);
+            return true;
+        } catch (error) {
+            this.logger.error(`Error deleting operator policy: ${error}`);
+            return false;
+        }
+    }
+
+    public async checkManagementPluginEnabled(): Promise<boolean> {
+        try {
+            const url = `${this.config.url}/api/overview`;
+
+            const response = await fetch(url, {
+                method: 'GET',
+                headers: {
+                    'Authorization': this.getAuthHeader()
+                }
+            });
+
+            return response.ok;
+        } catch (error) {
+            this.logger.warn(`Management plugin not accessible: ${error}`);
+            return false;
+        }
+    }
+}

src/core/message/RabbitMQMessage.ts

@@ -1,5 +1,5 @@
 import {Channel} from "amqplib";
-import {RunMQUtils} from "@src/core/utils/Utils";
+import {RunMQUtils} from "@src/core/utils/RunMQUtils";
 import {RabbitMQMessageProperties} from "@src/core/message/RabbitMQMessageProperties";
 import {AMQPMessage} from "@src/core/message/AmqpMessage";
 

src/core/utils/Utils.ts → src/core/utils/RunMQUtils.ts

@@ -16,4 +16,8 @@ export class RunMQUtils {
             throw new RunMQException(Exceptions.INVALID_MESSAGE_FORMAT, {});
         }
     }
+
+    public static escapeRegExp(string: string): string {
+        return string.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
+    }
 }

src/index.ts

@@ -6,5 +6,6 @@ export {
     RunMQProcessorConfiguration,
     MessageSchema,
     RunMQMessageContent,
-    RunMQMessageMetaContent
+    RunMQMessageMetaContent,
+    RabbitMQManagementConfig,
 } from "./types";

src/types/index.ts

@@ -26,6 +26,15 @@ export interface RunMQConnectionConfig {
      * Default is 10 attempts.
      */
     maxReconnectAttempts?: number;
+    /**
+     * Optional configuration for RabbitMQ management HTTP API.
+     * If provided, policies will be used for TTL instead of queue-level TTL.
+     */
+    management?: {
+        url: string;
+        username: string;
+        password: string;
+    };
 }
 
 export type SchemaFailureStrategy = 'dlq'
@@ -54,6 +63,32 @@ export interface RunMQProcessorConfiguration {
      * @see MessageSchema
      */
     messageSchema?: MessageSchema
+
+    /**
+     * Whether to use RabbitMQ policies for setting the attempts delay instead of queue-level TTL.
+     * Requires management configuration to be provided in RunMQConnectionConfig.
+     * Default is false.
+     *
+     * Recommended to use it for flexibility.
+     */
+    usePoliciesForDelay?: boolean;
+}
+
+export interface RabbitMQManagementConfig {
+    url: string;
+    username: string;
+    password: string;
+}
+
+export interface RabbitMQOperatorPolicy {
+    name: string;
+    pattern: string;
+    definition: {
+        "message-ttl"?: number;
+        [key: string]: any;
+    };
+    priority?: number;
+    "apply-to": "queues" | "exchanges" | "all";
 }
 
 export interface RunMQMessageContent<T = any> {