Configured customer broker

api-customer/src/controllers/mqtt_forwarder.py

@@ -0,0 +1,43 @@
+import json
+import time
+
+from src.services.mqtt_service import MQTTService
+
+
+class MQTTForwarder:
+
+    client_id : str
+    local_broker : MQTTService
+    central_broker : MQTTService
+
+    def __init__(self, client_id : str, local_mqtt: MQTTService, central_mqtt: MQTTService):
+        self.client_id = client_id
+        self.local_broker = local_mqtt
+        self.central_broker = central_mqtt
+
+    def start(self):
+        self.local_broker.subscribe("board-mate/", self.__forward)
+
+    def __forward(self, msg: str):
+        self.central_broker.publish(self.client_id, f"/board-mate/${self.client_id}/telemetry", msg)
+
+    """def start(self):
+        self.local.subscribe("board-mate/+/telemetry", self.handle_message)
+
+    def handle_message(self, topic: str, payload: str):
+        print(f"[FORWARD] {topic} -> central")
+
+        message = json.loads(payload)
+        message["source"] = "client-server"
+        message["forwarded_at"] = int(time.time())
+
+        self.central.publish(
+            client_id="client-forwarder",
+            topic=f"clients/{self.extract_client_id(topic)}/telemetry",
+            data=json.dumps(message),
+            qos=1
+        )
+
+    def extract_client_id(self, topic: str) -> str:
+        # ex: board-mate/pi-123/telemetry
+        return topic.split("/")[1]"""

api-customer/src/services/mqtt_service.py

@@ -0,0 +1,55 @@
+import json
+import time
+from typing import Callable
+
+import paho.mqtt.client as mqtt
+from paho.mqtt.client import Client
+
+
+class MQTTService:
+
+    client : Client
+
+    def __init__(self, address: str, port: int):
+        self.address = address
+        self.port = port
+        self.client = mqtt.Client()
+
+    def publish(self, client_id: str, topic: str, data: str, qos: int = 0):
+
+        try:
+            self.__connect()
+
+            payload = {
+                "timestamp": int(time.time()),
+                "data": data
+            }
+            result = self.client.publish(topic, json.dumps(payload), qos=qos)
+            result.wait_for_publish()
+
+        except Exception as e:
+            print("MQTT error:", e)
+            self.__disconnect()
+
+    def subscribe(self, topic: str, handler: Callable[[str], None]):
+        def on_message(client, userdata, msg):
+            handler(msg.payload.decode())
+
+        try:
+            self.__connect()
+            self.client.message_callback_add(topic, on_message)
+            self.client.subscribe(topic)
+        except Exception as e:
+            print("MQTT error:", e)
+            self.__disconnect()
+
+
+    def __connect(self):
+        if not self.client.is_connected():
+            self.client.connect(self.address, self.port)
+            self.client.loop_start()
+
+    def __disconnect(self):
+        if self.client.is_connected():
+            self.client.disconnect()
+            self.client.loop_stop()