remove gevent fixes #920

frigate/mqtt.py

@@ -1,7 +1,16 @@
+import json
 import logging
 import threading
+from wsgiref.simple_server import make_server
 
 import paho.mqtt.client as mqtt
+from ws4py.server.wsgirefserver import (
+    WebSocketWSGIHandler,
+    WebSocketWSGIRequestHandler,
+    WSGIServer,
+)
+from ws4py.server.wsgiutils import WebSocketWSGIApplication
+from ws4py.websocket import WebSocket
 
 from frigate.config import FrigateConfig
 
@@ -117,8 +126,15 @@ def create_mqtt_client(config: FrigateConfig, camera_metrics):
         )
 
     if not mqtt_config.tls_ca_certs is None:
-        if not mqtt_config.tls_client_cert is None and not mqtt_config.tls_client_key is None:
-            client.tls_set(mqtt_config.tls_ca_certs, mqtt_config.tls_client_cert, mqtt_config.tls_client_key)
+        if (
+            not mqtt_config.tls_client_cert is None
+            and not mqtt_config.tls_client_key is None
+        ):
+            client.tls_set(
+                mqtt_config.tls_ca_certs,
+                mqtt_config.tls_client_cert,
+                mqtt_config.tls_client_key,
+            )
         else:
             client.tls_set(mqtt_config.tls_ca_certs)
     if not mqtt_config.tls_insecure is None:
@@ -151,3 +167,79 @@ def create_mqtt_client(config: FrigateConfig, camera_metrics):
         )
 
     return client
+
+
+class MqttSocketRelay:
+    def __init__(self, mqtt_client, topic_prefix):
+        self.mqtt_client = mqtt_client
+        self.topic_prefix = topic_prefix
+
+    def start(self):
+        class MqttWebSocket(WebSocket):
+            topic_prefix = self.topic_prefix
+            mqtt_client = self.mqtt_client
+
+            def received_message(self, message):
+                try:
+                    json_message = json.loads(message.data.decode("utf-8"))
+                    json_message = {
+                        "topic": f"{self.topic_prefix}/{json_message['topic']}",
+                        "payload": json_message["payload"],
+                        "retain": json_message.get("retain", False),
+                    }
+                except Exception as e:
+                    logger.warning("Unable to parse websocket message as valid json.")
+                    return
+
+                logger.debug(
+                    f"Publishing mqtt message from websockets at {json_message['topic']}."
+                )
+                self.mqtt_client.publish(
+                    json_message["topic"],
+                    json_message["payload"],
+                    retain=json_message["retain"],
+                )
+
+        # start a websocket server on 5002
+        WebSocketWSGIHandler.http_version = "1.1"
+        self.websocket_server = make_server(
+            "127.0.0.1",
+            5002,
+            server_class=WSGIServer,
+            handler_class=WebSocketWSGIRequestHandler,
+            app=WebSocketWSGIApplication(handler_cls=MqttWebSocket),
+        )
+        self.websocket_server.initialize_websockets_manager()
+        self.websocket_thread = threading.Thread(
+            target=self.websocket_server.serve_forever
+        )
+
+        def send(client, userdata, message):
+            """Sends mqtt messages to clients."""
+            try:
+                logger.debug(f"Received mqtt message on {message.topic}.")
+                ws_message = json.dumps(
+                    {
+                        "topic": message.topic.replace(f"{self.topic_prefix}/", ""),
+                        "payload": message.payload.decode(),
+                    }
+                )
+            except Exception as e:
+                # if the payload can't be decoded don't relay to clients
+                logger.debug(
+                    f"MQTT payload for {message.topic} wasn't text. Skipping..."
+                )
+                return
+
+            self.websocket_server.manager.broadcast(ws_message)
+
+        self.mqtt_client.message_callback_add(f"{self.topic_prefix}/#", send)
+
+        self.websocket_thread.start()
+
+    def stop(self):
+        self.websocket_server.manager.close_all()
+        self.websocket_server.manager.stop()
+        self.websocket_server.manager.join()
+        self.websocket_server.shutdown()
+        self.websocket_thread.join()