feat: 🎨 receive MQTT message to create consumption marking

hydrometer/hydrometer.ino

@@ -23,6 +23,7 @@ const char* mqtt_server = "MQTT_SERVER_HERE";
 const char* mqtt_username = "MQTT_USERNAME";
 const char* mqtt_password = "MQTT_PASSWORD";
 const int mqtt_port = 1883;
+const char* mqtt_create_consumption_marking_topic = "createConsumptionMarking";
 
 const int HYROMETER_ID = 1;
 const char* HYDROMETER_PASSWORD = "example";
@@ -93,29 +94,19 @@ void sendDataToMQTT() {
   client.loop();
 
   String message = String(HYROMETER_ID) + "|" + HYDROMETER_PASSWORD + "|" + String(flowRate);
-  client.publish("createConsumptionMarking", message.c_str());
+  client.publish(mqtt_create_consumption_marking_topic, message.c_str());
 
   delay(5000);
 }
 
-void callbackMQTT(char* topic, byte* payload, unsigned int length) {
-  Serial.print("Message received [");
-  Serial.print(topic);
-  Serial.print("]: ");
-
-  for (int i = 0; i < length; i++) {
-    Serial.print((char)payload[i]);
-  }
-
-  Serial.println();
-}
+void callbackMQTT(char* topic, byte* payload, unsigned int length) {}
 
 void reconnect() {
   while (!client.connected()) {
     Serial.print("Connecting on MQTT...");
     if (client.connect("ESP8266Client", mqtt_username, mqtt_password)) {
       Serial.println("Connected!");
-      client.subscribe("createConsumptionMarking");
+      client.subscribe(mqtt_create_consumption_marking_topic);
     } else {
       Serial.print("Failed, rc=");
       Serial.print(client.state());

server/src/shared/infra/mqtt/index.ts

@@ -1,10 +1,14 @@
+import 'reflect-metadata'
 import mqtt from 'mqtt'
 
+import '@shared/containers'
+import { CreateConsumptionMarkingTopic } from './topics/CreateConsumptionMarkingTopic'
+
 const MQTT_BROKER_BASE_URL = process.env.MQTT_BROKER_BASE_URL ?? 'mqtt://localhost:3333'
 const MQTT_BROKER_USERNAME = process.env.MQTT_BROKER_USERNAME
 const MQTT_BROKER_PASSWORD = process.env.MQTT_BROKER_PASSWORD
 
-const MQTT_CREATE_CONSUMPTION_MARKING_TOPC = 'createConsumptionMarking'
+const MQTT_CREATE_CONSUMPTION_MARKING_TOPIC = 'createConsumptionMarking'
 
 const mqttClient = mqtt.connect(
   MQTT_BROKER_BASE_URL,
@@ -15,14 +19,22 @@ const mqttClient = mqtt.connect(
 )
 
 mqttClient.on('connect', async () => {
-  console.log('Connnected on MQTT')
-  mqttClient.subscribe(MQTT_CREATE_CONSUMPTION_MARKING_TOPC)
+  console.log('[Connnected on MQTT]')
+  mqttClient.subscribe(MQTT_CREATE_CONSUMPTION_MARKING_TOPIC)
+})
+
+mqttClient.on('error', async (error) => {
+  console.log('[MQTT Error]: ' + error)
 })
 
 mqttClient.on('message', async (topic, message) => {
-  console.log('Message received [' + topic + ']: ' + message.toString());
-});
+  console.log('[Message received in the topic ' + topic + ']: ' + message.toString())
 
-mqttClient.on('error', async (error) => {
-  console.log('Error: ' + error);
+  if (topic === MQTT_CREATE_CONSUMPTION_MARKING_TOPIC) {
+    const messageString = message.toString()
+
+    const createConsumptionMarkingTopic = new CreateConsumptionMarkingTopic()
+
+    await createConsumptionMarkingTopic.receiveMessage(messageString)
+  }
 })

server/src/shared/infra/mqtt/topics/CreateConsumptionMarkingTopic.ts

@@ -0,0 +1,17 @@
+import { container } from 'tsyringe'
+
+import { CreateConsumptionMarkingUseCase } from '@modules/consumptionMarkings/useCases/createConsumptionMarking/CreateConsumptionMarkingUseCase'
+
+export class CreateConsumptionMarkingTopic {
+  public async receiveMessage(message: string) {
+    const [hydrometer_id, hydrometer_password, consumption] = message.split('|')
+
+    const createConsumptionMarkingUseCase = container.resolve(CreateConsumptionMarkingUseCase)
+
+    await createConsumptionMarkingUseCase.execute({
+      consumption: Number(consumption),
+      hydrometer_id: Number(hydrometer_id),
+      hydrometer_password,
+    })
+  }
+}