NestJS RabbitMQ Producer Module

A reusable NestJS module for producing messages to RabbitMQ with built-in message validation.

Installation

npm install @lneimark/nestjs-rabbitmq-producer

Usage

1. Import the module in your NestJS application:

import { Module } from '@nestjs/common';
import { ProducerModule } from '@lneimark/nestjs-rabbitmq-producer';

@Module({
	imports: [ProducerModule],
})
export class AppModule {}

2. Configure your RabbitMQ connection in your .env file:

RMQ_USER=your_user
RMQ_PASS=your_password
RMQ_HOST=your_host
RMQ_PORT=5672
RMQ_QUEUE=your_queue

3. Use the ProducerService in your services:

import { Injectable } from '@nestjs/common';
import { ProducerService } from '@lneimark/nestjs-rabbitmq-producer';

@Injectable()
export class YourService {
	constructor(private readonly producerService: ProducerService) {}

	async sendStrategyUpdate() {
		const message = {
			"SKU": "MLB123456789",
			"MLA": "MLB987654321",
			"date": "2024-03-20T10:30:00Z",
			"strategy_id": "STR-2024-001",
			"strategy_name": "Premium Pricing Strategy",
			"strategy_type": "dynamic_pricing",
			"initial_quantity_stock": 100,
			"current_margin": 25.5,
			"price": 199.99,
			"minimum_margin": 15.0,
			"minimum_price": 149.99
		} ;
		try {
			await this.producerService.produceStrategy(message);
		} catch(error) {
			...
		}
	}
}

Strategy update message format

	"SKU": "MLB123456789",
	"MLA": "MLB987654321",
	"date": "2024-03-20T10:30:00Z",
	"strategy_id": "STR-2024-001", -> Not mandatory
	"strategy_name": "Premium Pricing Strategy", -> Not mandatory
	"strategy_type": "dynamic_pricing", -> Not mandatory
	"initial_quantity_stock": 100, -> Not mandatory
	"current_margin": 25.5,
	"price": 199.99,
	"minimum_margin": 15.0,
	"minimum_price": 149.99

Testing locally

To test this service locally you need to have a rabbitmq connection running on the port stablished on the .env

for running this on Docker here is the image

FROM rabbitmq:3-management

# Habilitar plugins
RUN rabbitmq-plugins enable --offline rabbitmq_management

# Copiar el archivo de definiciones al contenedor
COPY rabbitmq_definitions.json /etc/rabbitmq/definitions.json

# Variables de entorno para cargar definiciones
ENV RABBITMQ_DEFINITIONS_FILE=/etc/rabbitmq/definitions.json
ENV RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_management load_definitions "/etc/rabbitmq/definitions.json"'

# Crear el usuario por defecto
ENV RABBITMQ_DEFAULT_USER=admin
ENV RABBITMQ_DEFAULT_PASS=admin

# Exponer puertos necesarios
EXPOSE 5672 15672

# Comando por defecto
CMD ["rabbitmq-server"]

Here is the rabbitmq_definitions.json that needs to be in the same folder as the image

{
	"rabbit_version": "3.13.7",
	"rabbitmq_version": "3.13.7",
	"product_name": "RabbitMQ",
	"product_version": "3.13.7",
	"users": [
		{
			"name": "API",
			"password_hash": "DKUBDF64cx7a/RnhCfUUEVAXWxhJqo4Fpl5zmvgglE8ujdl7",
			"hashing_algorithm": "rabbit_password_hashing_sha256",
			"tags": ["publisher"],
			"limits": {}
		},
		{
			"name": "admin",
			"password_hash": "DKUBDF64cx7a/RnhCfUUEVAXWxhJqo4Fpl5zmvgglE8ujdl7",
			"hashing_algorithm": "rabbit_password_hashing_sha256",
			"tags": ["administrator"],
			"limits": {}
		},
		{
			"name": "ADS_ETL",
			"password_hash": "DKUBDF64cx7a/RnhCfUUEVAXWxhJqo4Fpl5zmvgglE8ujdl7",
			"hashing_algorithm": "rabbit_password_hashing_sha256",
			"tags": ["reader"],
			"limits": {}
		},
		{
			"name": "MICROSERVICES",
			"password_hash": "DKUBDF64cx7a/RnhCfUUEVAXWxhJqo4Fpl5zmvgglE8ujdl7",
			"hashing_algorithm": "rabbit_password_hashing_sha256",
			"tags": ["publisher"],
			"limits": {}
		},
		{
			"name": "CMV_ETL",
			"password_hash": "DKUBDF64cx7a/RnhCfUUEVAXWxhJqo4Fpl5zmvgglE8ujdl7",
			"hashing_algorithm": "rabbit_password_hashing_sha256",
			"tags": ["reader"],
			"limits": {}
		},
		{
			"name": "STOCK_ETL",
			"password_hash": "DKUBDF64cx7a/RnhCfUUEVAXWxhJqo4Fpl5zmvgglE8ujdl7",
			"hashing_algorithm": "rabbit_password_hashing_sha256",
			"tags": ["reader"],
			"limits": {}
		}
	],
	"vhosts": [
		{
			"name": "/",
			"description": "Default virtual host",
			"tags": [],
			"metadata": {
				"description": "Default virtual host",
				"tags": []
			}
		}
	],
	"permissions": [
		{
			"user": "API",
			"vhost": "/",
			"configure": "",
			"write": ".*",
			"read": ""
		},
		{
			"user": "ADS_ETL",
			"vhost": "/",
			"configure": "",
			"write": "^(ads_retry_q|amq\\.default)$",
			"read": "^(ads_queue|ads_retry_q)$"
		},
		{
			"user": "CMV_ETL",
			"vhost": "/",
			"configure": "",
			"write": "^(company_etl_retry_q|amq\\.default)$",
			"read": "^(cmv_queue|company_etl_retry_q)$"
		},
		{
			"user": "admin",
			"vhost": "/",
			"configure": ".*",
			"write": ".*",
			"read": ".*"
		},
		{
			"user": "MICROSERVICES",
			"vhost": "/",
			"configure": "",
			"write": ".*",
			"read": ""
		},
		{
			"user": "STOCK_ETL",
			"vhost": "/",
			"configure": "",
			"write": "^(stock_etl_retry_q|amq\\.default)$",
			"read": "^(stock_queue|stock_retry_q)$"
		}
	],
	"topic_permissions": [],
	"parameters": [],
	"global_parameters": [
		{
			"name": "internal_cluster_id",
			"value": "rabbitmq-cluster-id-SD8--gKdPUGLlBMEKJ7K1w"
		}
	],
	"policies": [],
	"queues": [
		{
			"name": "ads_queue",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {}
		},
		{
			"name": "cmv_queue",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {}
		},
		{
			"name": "stock_queue",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {}
		},
		{
			"name": "cmv_retry_q",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {
				"x-dead-letter-exchange": "",
				"x-dead-letter-routing-key": "cmv_queue",
				"x-message-ttl": 300000
			}
		},
		{
			"name": "ads_retry_q",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {
				"x-dead-letter-exchange": "",
				"x-dead-letter-routing-key": "ads_queue",
				"x-message-ttl": 300000
			}
		},
		{
			"name": "stock_retry_q",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {
				"x-dead-letter-exchange": "",
				"x-dead-letter-routing-key": "stock_queue",
				"x-message-ttl": 300000
			}
		},
		{
			"name": "strategy_real_time_queue",
			"vhost": "/",
			"durable": true,
			"auto_delete": false,
			"arguments": {}
		}
	],
	"exchanges": [],
	"bindings": []
}

For running this, use this command

docker build -t rabbitmq-custom -f Dockerfile .

docker run -d --name rabbitmq-custom -p 5672:5672 -p 15672:15672 rabbitmq-custom

remember: if running locally, the host is localhost on the .env