test rabbitmq email service

backend/.env.example

@@ -1,6 +1,5 @@
 PORT=3000
 NODE_ENV=development
-RABBITMQ=localhost
 
 DB_HOST=localhost
 DB_PORT=3306
@@ -18,4 +17,12 @@ MQTT_USER=emqx
 MQTT_PASSWORD=public
 
 VAPID_PUBLIC_KEY=<VAPID_PUBLIC_KEY>
-VAPID_PRIVATE_KEY=<VAPID_PRIVATE_KEY>
+VAPID_PRIVATE_KEY=<VAPID_PRIVATE_KEY>
+
+AQI_API_TOKEN=4ae4ef041a1ee5684232e273986a11d9983b5093
+
+PREDICTION_SERVICE_URL=http://localhost:8081
+
+RABBITMQ=amqp://admin:admin@localhost:5672
+RABBITMQ_MAILSERVICE_EXCHANGE_NAME=EmailExchange
+RABBITMQ_MAILSERVICE_QUEUE_NAME=MailServiceQueue

backend/src/app.js

@@ -13,6 +13,9 @@ const { errorHandler } = require("./middlewares/errors");
 const ApiError = require("./utils/ApiError");
 const httpStatus = require("http-status");
 
+const Producer = require("./rabbitmq/producer");
+const producer = new Producer();
+
 const start = async (agenda) => {
 	const app = express();
 
@@ -43,6 +46,11 @@ const start = async (agenda) => {
 const apiRoutes = (app, io, mqtt) => {
 	app.use("/v1", v1Route);
 
+	app.post("/amqp", async (req, res, next) => {
+		await producer.publishEmailMessage(req.body.mailList);
+		res.send("Sent");
+	});
+
 	app.get("/", async (req, res) => {
 		res.set("Content-Type", "text/html");
 		res.send(

backend/src/config/config.js

@@ -26,6 +26,9 @@ const envVarsSchema = Joi.object()
 		VAPID_PRIVATE_KEY: Joi.string(),
 		AQI_API_TOKEN: Joi.string(),
 		PREDICTION_SERVICE_URL: Joi.string(),
+		RABBITMQ: Joi.string(),
+		RABBITMQ_MAILSERVICE_EXCHANGE_NAME: Joi.string(),
+		RABBITMQ_MAILSERVICE_QUEUE_NAME: Joi.string(),
 	})
 	.unknown();
 
@@ -62,4 +65,9 @@ module.exports = {
 	vapidPrivateKey: envVars.VAPID_PRIVATE_KEY,
 	aqiApiToken: envVars.AQI_API_TOKEN,
 	predictionServiceUrl: envVars.PREDICTION_SERVICE_URL,
+	rabbitmq: {
+		url: envVars.RABBITMQ,
+		mailServiceExchangeName: envVars.RABBITMQ_MAILSERVICE_EXCHANGE_NAME,
+		mailServiceQueueName: envVars.RABBITMQ_MAILSERVICE_QUEUE_NAME,
+	},
 };

backend/src/rabbitmq/producer.js

@@ -0,0 +1,39 @@
+const amqp = require("amqplib");
+const config = require("../config/config");
+const logger = require("../config/logger");
+
+class Producer {
+	channel;
+
+	async createChannel() {
+		const connection = await amqp.connect(config.rabbitmq.url);
+		this.channel = await connection.createChannel();
+	}
+
+	async publishEmailMessage(message, routingKey = "MailService") {
+		if (!this.channel) {
+			await this.createChannel();
+		}
+
+		const mailServiceExchangeName = config.rabbitmq.mailServiceExchangeName;
+		await this.channel.assertExchange(mailServiceExchangeName, "direct");
+
+		const logDetails = {
+			logType: routingKey,
+			message: message,
+			dateTime: new Date(),
+		};
+
+		await this.channel.publish(
+			mailServiceExchangeName,
+			routingKey,
+			Buffer.from(JSON.stringify(logDetails))
+		);
+
+		logger.info(
+			`The new ${routingKey} is sent to exchange ${mailServiceExchangeName}`
+		);
+	}
+}
+
+module.exports = Producer;

docker-compose.yml

@@ -129,7 +129,6 @@ services:
       - "15672:15672"
     volumes:
       - ./airchecker/rabbitmq:/var/lib/rabbitmq
-      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
     networks:
       - node-network
 

email-service/app.js

@@ -1,7 +1,38 @@
+const config = require("./config.json");
+const logger = require("./logger");
+const amqp = require("amqplib");
+
 const sendMail = require("./transporter");
 
-const mailList = ["nguyenduckhai8101@gmail.com", "19521658@gm.uit.edu.vn"];
+async function consumeMessages() {
+	const connection = await amqp.connect(config.amqp.url);
+	const channel = await connection.createChannel();
+
+	await channel.assertExchange(config.amqp.exchangeName, "direct");
+
+	const q = await channel.assertQueue(config.amqp.queueName);
+
+	await channel.bindQueue(
+		q.queue,
+		config.amqp.exchangeName,
+		config.amqp.bindingKey
+	);
+
+	channel.consume(q.queue, async (msg) => {
+		const data = JSON.parse(msg.content);
+		logger.info(
+			`Receive message from exchange [${
+				config.amqp.exchangeName
+			}], data ${JSON.stringify(data)}`
+		);
+
+		const { message: mailList } = data;
+		console.log(mailList);
+
+		await sendMail(mailList);
+
+		channel.ack(msg);
+	});
+}
 
-(async () => {
-	await sendMail(mailList);
-})();
+consumeMessages();

email-service/config.json

@@ -1,6 +1,10 @@
 {
-	"amqp": "amqp://localhost",
-	"queue": "nodemailer-amqp",
+	"amqp": {
+		"exchangeName": "EmailExchange",
+		"queueName": "MailServiceQueue",
+		"bindingKey": "MailService",
+		"url": "amqp://admin:admin@localhost:5672"
+	},
 
 	"smtp_server": {
 		"port": 465,

email-service/logger.js

@@ -18,7 +18,7 @@ const logger = winston.createLogger({
 		winston.format.errors({ stack: true }),
 		winston.format.json()
 	),
-	defaultMeta: { application: "mail_service" },
+	defaultMeta: { application: "email_service" },
 	transports: [
 		new winston.transports.Console({
 			format: winston.format.simple(),