changed protocol from AMQP to MQTT in rabbitmq client + changes in callback fun

fjpacheco · violetaperezandrade · fjpacheco · commit 00c6b1f01609 · 2024-06-11T23:18:07.000-03:00
Co-authored-by: violeta &lt;violetaperezandrade@users.noreply.github.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -5,3 +5,4 @@ __pycache__/
 *.lock
 .DS_Store
 vscode/
+/venv
diff --git a/app/main_rabbitmq.py b/app/main_rabbitmq.py
@@ -10,13 +10,15 @@ def main():
     firebase_credentials = os.environ.get('FIREBASE_CREDENTIALS')
     cred = credentials.Certificate(json.loads(firebase_credentials))
     initialize_app(cred)
+    
     logger = logging.getLogger("rabbitmq_consumer")
-    # DEBUG, INFO, WARNING, ERROR, CRITICAL
-    logging_level = os.environ.get("LOGGING_LEVEL")
-    queue_name = os.environ.get("QUEUE_NAME")
+    logging_level = os.environ.get("LOGGING_LEVEL", "info")
     initialize_log(logging_level)
+    
     loop = asyncio.get_event_loop()
-    consumer = Consumer(queue_name)
+    
+    topic_name = os.environ.get("MQTT_TOPIC", "measurements")
+    consumer = Consumer(topic_name)
     consumer.run()
     try:
         loop.run_forever()
diff --git a/app/service/common/middleware.py b/app/service/common/middleware.py
@@ -1,51 +1,76 @@
-import pika
 import os
+import time
+import logging
+import paho.mqtt.client as mqtt
 
+logger = logging.getLogger("rabbitmq_consumer")
+logging.getLogger("pika").setLevel(logging.WARNING)
+
+FIRST_RECONNECT_DELAY = 1
+RECONNECT_RATE = 2
+MAX_RECONNECT_COUNT = 12
+MAX_RECONNECT_DELAY = 60
 
-class Middleware:
 
+class Middleware:
     def __init__(self):
-        self._connection = pika.BlockingConnection(
-            parameters=pika.URLParameters(
-                os.environ.get("CLOUDAMQP_URL", "amqp://rabbitmq:5672")))
-        self._channel = self._connection.channel()
-        self._exit = False
-        self._remake = False
-
-    def create_queue(self, queue_name):
-        self._channel.queue_declare(queue=queue_name)
-
-    def _setup_message_consumption(self, queue_name, user_function):
-        self._channel.basic_consume(queue=queue_name,
-                                    on_message_callback=lambda channel,
-                                    method, properties, body:
-                                    (user_function(body),
-                                     channel.basic_ack
-                                     (delivery_tag=method.delivery_tag),
-                                     self._verify_connection_end()))
-        self._channel.start_consuming()
-
-    def _verify_connection_end(self):
-        if self._exit:
-            self._channel.close()
-            if self._remake:
-                self._exit = False
-                self._channel = self._connection.channel()
-
-    def finish(self, open_new_channel=False):
-        self._exit = True
-        self._remake = open_new_channel
-
-    # Work queue methods
-    def listen_on(self, queue_name, user_function):
-        self.create_queue(queue_name)
-        self._channel.basic_qos(prefetch_count=30)
-        self._setup_message_consumption(queue_name, user_function)
-
-    def send_message(self, queue_name, message):
-        self._channel.basic_publish(exchange='',
-                                    routing_key=queue_name,
-                                    body=message)
+        self._client = mqtt.Client()
+        self._client.username_pw_set(os.environ.get("MQTT_USERNAME"),
+                                     os.environ.get("MQTT_PASSWORD"))
+        self._topic = None
+
+    def _on_connect(self, client, userdata, flags, rc):
+        if rc == 0:
+            logger.info("Connected successfully")
+            if self._topic is not None:
+                self._client.subscribe(self._topic)
+                logger.info(f"[x] Subscribed to {self._topic}")
+        else:
+            logger.info(f"Connect failed with code {rc}")
+
+    def _on_disconnect(client, userdata, rc):
+        logger.info("Disconnected with result code: %s", rc)
+        reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY
+        while reconnect_count < MAX_RECONNECT_COUNT:
+            logger.info("Reconnecting in %d seconds...", reconnect_delay)
+            time.sleep(reconnect_delay)
+
+            try:
+                client.reconnect()
+                logger.info("Reconnected successfully!")
+                return
+            except Exception as err:
+                logger.error("%s. Reconnect failed. Retrying...", err)
+
+            reconnect_delay *= RECONNECT_RATE
+            reconnect_delay = min(reconnect_delay, MAX_RECONNECT_DELAY)
+            reconnect_count += 1
+        logger.info("Reconnect failed after %s attempts. Exiting...",
+                    reconnect_count)
+
+    def listen_on(self, user_function, topic):
+        self._topic = topic
+        self._client.on_connect = self._on_connect
+        self._client.on_message = user_function
+        self._client.on_disconnect = self._on_disconnect
+
+    def send_message(self, topic, msg):
+        result = self._client.publish(topic, msg)
+        status = result[0]
+        if status == 0:
+            logger.info(f"Send `{msg}` to topic `{topic}`")
+        else:
+            logger.info(f"Failed to send message to topic {topic}")
 
     def __del__(self):
-        self._connection.close()
+        self._client.disconnect()
+
+    def connect(self):
+        self._client.connect(os.environ.get("MQTT_HOST"),
+                             int(os.environ.get("MQTT_PORT")))
+
+    def run(self):
+        self._client.loop_start()
+        
+    def finish(self):
+        self._client.disconnect()
diff --git a/app/service/rabbitmq/consumer.py b/app/service/rabbitmq/consumer.py
@@ -1,22 +1,22 @@
-import asyncio
 import json
-import pydantic
 import logging
-
+import asyncio
+import pydantic
+from os import environ
 from pydantic import ValidationError
 from sqlalchemy import MetaData
 from sqlalchemy.orm import Session
 from sqlalchemy import create_engine
 from sqlalchemy.ext.declarative import declarative_base
-from os import environ
 from firebase_admin import messaging
 
 from external.Users import UsersService
 from ..common.middleware import Middleware
 from database.models.measurement import Measurement
 from database.database import SQLAlchemyClient
-from resources.parser import apply_rules
+from resources.parser_rules import apply_rules
 from schemas.measurement import MeasurementReadingSchema
+from resources.parser_messages import parse_message
 from exceptions.logger_messages import LoggerMessages
 from exceptions.invalid_insertion import InvalidInsertionError
 from exceptions.deviating_parameters import DeviatedParametersError
@@ -25,25 +25,27 @@
 
 
 Base = declarative_base(
-    metadata=MetaData(schema=environ.get("POSTGRES_SCHEMA", "measurements_service"))
+    metadata=MetaData(
+        schema=environ.get("POSTGRES_SCHEMA", "measurements_service")
+        )
 )
 logger = logging.getLogger("rabbitmq_consumer")
-logging.getLogger("pika").setLevel(logging.WARNING)
 dbUrl = environ.get("DATABASE_URL").replace("postgres://", "postgresql://", 1)
 engine = create_engine(dbUrl, echo=True, future=True)
 session = Session(engine)
 
-
 class Consumer:
-    def __init__(self, queue_name):
-        self.__queue_name = queue_name
+    def __init__(self, topic):
         self.__middleware = Middleware()
         self.__sqlAlchemyClient = SQLAlchemyClient()
         self.__users = UsersService()
+        self.__last_measurements = {}
+        self.__topic = topic
 
     def run(self):
-        self.__middleware.create_queue(self.__queue_name)
-        self.__middleware.listen_on(self.__queue_name, self.__callback)
+        self.__middleware.listen_on(self.__callback, self.__topic)
+        self.__middleware.connect()
+        self.__middleware.run()
 
     def obtain_device_plant(self, measurement_from_rabbit):
         try:
@@ -54,12 +56,13 @@ def obtain_device_plant(self, measurement_from_rabbit):
             return dp
         except Exception as err:
             logger.error(f"{err} - {type(err)}")
-            raise RowNotFoundError(measurement_from_rabbit.id_device, "DEVICE_PLANT")
+            raise RowNotFoundError(measurement_from_rabbit.id_device,
+                                   "DEVICE_PLANT")
 
     async def obtain_user(self, user_id):
         return await self.__users.get_user(user_id)
 
-    def check_package(self, measurement):
+    def check_package(self, measurement: MeasurementReadingSchema):
         empty_values = []
         if measurement.temperature is None:
             empty_values.append("temperature")
@@ -70,7 +73,9 @@ def check_package(self, measurement):
         if measurement.light is None:
             empty_values.append("light")
 
-        if measurement.humidity is None:
+        # Nuestro sensor no mide humedad del ambiente!
+        if measurement.humidity is None and not measurement.id_device.\
+                startswith("sensor_"):
             empty_values.append("humidity")
 
         if measurement.id_device is None:
@@ -113,24 +118,24 @@ def generate_notification_body(self, error):
     async def send_notification(self, id_user, measurement, error, details):
         device_plant = self.obtain_device_plant(measurement)
 
-        user = await self.obtain_user(device_plant.id_user)
-
-        notification_body = self.generate_notification_body(error)
-
         try:
+            user = await self.obtain_user(device_plant.id_user)
+            notification_body = self.generate_notification_body(error)
             if user.device_token is not None:
                 message = messaging.Message(
-                    notification=messaging.Notification(title="Estado de tu planta",
-                                                        body=notification_body),
-                    token=user.device_token)
+                    notification=messaging.Notification(
+                        title="Estado de tu planta", body=notification_body
+                    ),
+                    token=user.device_token,
+                )
                 messaging.send(message)
             logger.info(LoggerMessages.USER_NOTIFIED.format(id_user))
 
         except Exception as e:
             logger.info(e)
             pass
 
-    def apply_rules(self, measurement,  device_plant):
+    def apply_rules(self, measurement: MeasurementReadingSchema, device_plant):
         deviated_parameters = apply_rules(measurement, device_plant.plant_type)
         if deviated_parameters.hasDeviations():
             raise DeviatedParametersError(deviated_parameters)
@@ -156,16 +161,25 @@ def save_measurement(self, measurement_from_rabbit, device_plant):
         except Exception as err:
             logger.error(f"{err} - {type(err)}")
             self.__sqlAlchemyClient.rollback()
-            raise InvalidInsertionError(measurement_from_rabbit, "MEAUSUREMENT")
+            raise InvalidInsertionError(measurement_from_rabbit,
+                                        "MEAUSUREMENT")
 
-    def __callback(self, body):
+    def __callback(self, client, userdata, msg):
         device_plant = None
-        measurement = None
+        measurement, body = parse_message(self.__last_measurements, msg)
+        if measurement is None:
+            return
+
+        # 1) flood de notificaciones !!
+        # 2) light!!!!
+        # 3) humedad! no enviamos notificacion cuando llega none...??? [OK]
+        # 4) convertir int a float en simulador y microservice [OK]
 
+        # 5) refactor decode_body [OK]
         try:
-            measurement = MeasurementReadingSchema(**json.loads(body))
             logger.info(
-                LoggerMessages.NEW_PACKAGE_RECEIVED.format(measurement.id_device)
+                LoggerMessages.NEW_PACKAGE_RECEIVED
+                .format(measurement.id_device)
             )
             logger.debug(LoggerMessages.PACKAGE_DETAIL.format(body))
 
@@ -181,7 +195,8 @@ def __callback(self, body):
             logger.debug(LoggerMessages.ERROR_DETAILS.format(err, body))
         except RowNotFoundError as err:
             logger.warn(
-                LoggerMessages.ROW_NOT_FOUND.format(err.primary_key, err.name_table)
+                LoggerMessages.ROW_NOT_FOUND.format(err.primary_key,
+                                                    err.name_table)
             )
             logger.debug(LoggerMessages.ERROR_DETAILS.format(err, body))
 
@@ -195,7 +210,8 @@ def __callback(self, body):
             logger.warn(LoggerMessages.DEVIATING_PARAMETERS)
             logger.debug(LoggerMessages.ERROR_DETAILS.format(err, body))
 
-            asyncio.run(self.send_notification(device_plant.id_user, measurement,
+            asyncio.run(self.send_notification(device_plant.id_user,
+                                               measurement,
                                                err, body))
 
         if device_plant is not None and measurement is not None:
diff --git a/docker-compose.yaml b/docker-compose.yaml
@@ -41,38 +41,38 @@ services:
     networks:
       - common_network
 
-  # sql:
-  #   build:
-  #     context: ./app/service/sql
-  #     dockerfile: Dockerfile
-  #   env_file:
-  #     - .env
-  #   ports:
-  #     - "5432:5432"
-  #   volumes:
-  #     - ./app/service/sql/init_table_device_plant.sql:/docker-entrypoint-initdb.d/init_table_device_plant.sql
-  #     - ./app/service/sql/init_table_measurements.sql:/docker-entrypoint-initdb.d/init_table_measurements.sql
-  #   healthcheck:
-  #     test:
-  #       [
-  #         "CMD",
-  #         "pg_isready",
-  #         "-h",
-  #         "localhost",
-  #         "-p",
-  #         "5432",
-  #         "-q",
-  #         "-U",
-  #         "${POSTGRES_USER}",
-  #         "-d",
-  #         "${POSTGRES_DB}"
-  #       ]
-  #     interval: 5s
-  #     timeout: 3s
-  #     retries: 10
-  #     start_period: 50s
-  #   networks:
-  #     - common_network
+  sql:
+    build:
+      context: ./app/service/sql
+      dockerfile: Dockerfile
+    env_file:
+      - .env
+    ports:
+      - "5432:5432"
+    volumes:
+      - ./app/service/sql/init_table_device_plant.sql:/docker-entrypoint-initdb.d/init_table_device_plant.sql
+      - ./app/service/sql/init_table_measurements.sql:/docker-entrypoint-initdb.d/init_table_measurements.sql
+    healthcheck:
+      test:
+        [
+          "CMD",
+          "pg_isready",
+          "-h",
+          "localhost",
+          "-p",
+          "5432",
+          "-q",
+          "-U",
+          "${POSTGRES_USER}",
+          "-d",
+          "${POSTGRES_DB}"
+        ]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+      start_period: 50s
+    networks:
+      - common_network
       
 networks:
   common_network:
diff --git a/requirements.txt b/requirements.txt

-Original file line number
+Diff line change
 *.lock
 .DS_Store
 vscode/
 +/venv