external_openwb: durchgehende MQTT-Verbindung statt Neuverbindung pro Zyklus

packages/helpermodules/broker.py

@@ -1,11 +1,21 @@
 import datetime
+import errno
 import logging
 import paho.mqtt.client as mqtt
+import threading
 import time
-from typing import Callable
+from typing import Any, Callable, Dict, List, Tuple
+
+from helpermodules.utils.topic_parser import decode_payload
 
 log = logging.getLogger(__name__)
 
+# Zeitfenster nach dem ersten Verbindungsversuch, in dem "noch nicht verbunden" nicht als Fehler
+# gilt: connect_async()/loop_start() verbinden im Hintergrund, is_connected() ist direkt danach
+# so gut wie immer noch False. Ohne Kulanzfrist würde der allererste publish() an einen neuen Host
+# nach jedem Neustart praktisch garantiert einen falschen Fehler auslösen.
+INITIAL_CONNECT_GRACE = 5.0
+
 
 def get_name_suffix() -> str:
     with open('/proc/cpuinfo', 'r') as f:
@@ -45,6 +55,122 @@ def disconnect(self) -> None:
         log.info(f"Verbindung von Client {self.name} geschlossen.")
 
 
+class PersistentBrokerClient:
+    """Lang lebende, automatisch wiederverbindende MQTT-Verbindung zu einem Host (typischerweise ein
+    Secondary-openWB), die über mehrere Regelzyklen hinweg wiederverwendet wird.
+
+    Ersetzt für publish/subscribe zu externen Hosts das bisherige Muster "pro Aufruf neu verbinden,
+    kurz warten, wieder trennen" (paho publish.single() bzw. BrokerClient.start_finite_loop()): jeder
+    dieser Auf- und Abbauten ist ein eigener TCP+MQTT-Handshake, und mehrere davon pro Regelzyklus
+    (z.B. 5x publish + 1x subscribe für einen externen Ladepunkt) mussten bisher gemeinsam in das enge
+    Timeout von control_interval/3 passen - schon normaler Netzwerk-Jitter zur Gegenstelle konnte das
+    reißen. Mit einer dauerhaft offenen, im Hintergrund automatisch wiederverbundenen Verbindung fällt
+    dieser Auf-/Abbau pro Zyklus komplett weg.
+    """
+
+    def __init__(self, host: str, port: int) -> None:
+        self.host = host
+        self.port = port
+        self._lock = threading.Lock()
+        self._subscribed_topics: set = set()
+        self._received_topics: Dict[str, Any] = {}
+        # eigenes Event je wartendem Aufrufer statt eines gemeinsamen: bei zwei Ladepunkten am selben
+        # Host (duo_num 0/1, gemeinsame Verbindung) laufen mehrere wait_for_fresh_data()-Aufrufe
+        # gleichzeitig auf demselben PersistentBrokerClient - ein gemeinsames Event hätte dazu geführt,
+        # dass ein Aufrufer das Event im eigenen clear() dem anderen wegnimmt (verpasster Wakeup).
+        self._waiters: List[threading.Event] = []
+        self._ever_connected = False
+        self._created_at = time.monotonic()
+        self.client = mqtt.Client(f"openWB-persistent-{host}-{port}-{get_name_suffix()}")
+        self.client.on_connect = self._on_connect
+        self.client.on_message = self._on_message
+        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
+        try:
+            self.client.connect_async(host, port)
+            self.client.loop_start()
+        except Exception:
+            log.exception(f"Fehler beim Verbindungsaufbau zu {host}:{port}")
+
+    def _on_connect(self, client, userdata, flags, rc) -> None:
+        self._ever_connected = True
+        with self._lock:
+            topics = list(self._subscribed_topics)
+        for topic in topics:
+            client.subscribe(topic)
+
+    def _on_message(self, client, userdata, message) -> None:
+        with self._lock:
+            self._received_topics[message.topic] = decode_payload(message.payload)
+            waiters = list(self._waiters)
+        for waiter in waiters:
+            waiter.set()
+
+    def publish(self, topic: str, payload, qos: int = 0, retain: bool = True) -> None:
+        # anders als paho publish.single() wirft Client.publish() bei fehlender Verbindung keine Exception,
+        # sondern liefert nur einen Fehlercode zurueck - hier bewusst in eine Exception uebersetzt, damit
+        # Aufrufer (insbesondere ErrorTimerContext/client_error_context) eine nicht erreichbare Gegenstelle
+        # weiterhin wie bisher als Fehler erkennen und nach 60s die Sicherheitsabschaltung greifen kann.
+        if not self.client.is_connected():
+            if not self._ever_connected and time.monotonic() - self._created_at < INITIAL_CONNECT_GRACE:
+                # Erstverbindung läuft im Hintergrund noch (siehe INITIAL_CONNECT_GRACE weiter oben) -
+                # das ist kein Fehlerfall, Nachricht wird verworfen, naechster Zyklus greift wieder.
+                return
+            # errno.EHOSTUNREACH statt nur einer Nachricht, damit der bestehende OSError-Handler
+            # (helpermodules/exceptions/os.py) die schon vorhandene, uebersetzte Meldung "Die Verbindung
+            # zum Host ist fehlgeschlagen..." verwendet, statt auf "Unbekannter Fehler" zurueckzufallen -
+            # der greift nur ueber e.errno/e.strerror, eine reine Textnachricht wuerde er ignorieren.
+            raise ConnectionError(errno.EHOSTUNREACH, f"Keine Verbindung zu {self.host}:{self.port}")
+        self.client.publish(topic, payload, qos=qos, retain=retain)
+
+    def ensure_subscribed(self, topic_filter: str) -> None:
+        with self._lock:
+            if topic_filter in self._subscribed_topics:
+                return
+            self._subscribed_topics.add(topic_filter)
+        self.client.subscribe(topic_filter)
+
+    def wait_for_fresh_data(self, max_wait: float = 2.0, settle: float = 0.1) -> Dict[str, Any]:
+        """Wartet auf mindestens eine neue Nachricht und danach auf eine kurze Ruhephase ohne weitere
+        Nachrichten (Sammeln des restlichen Nachrichten-Schwalls, z.B. alle retained Topics nach einem
+        Reconnect), insgesamt begrenzt auf `max_wait`. Liefert bei mindestens einer neuen Nachricht den
+        gesamten aktuellen Stand aller je empfangenen Topics zurück (nicht nur die seit dem Aufruf neu
+        eingetroffenen) - sonst ein leeres Dict, damit Aufrufer "keine aktuellen Daten diesen Zyklus"
+        weiterhin wie bisher von "Daten vorhanden" unterscheiden können (z.B. für Fault-State-Meldungen)."""
+        waiter = threading.Event()
+        with self._lock:
+            self._waiters.append(waiter)
+        try:
+            deadline = time.monotonic() + max_wait
+            if not waiter.wait(timeout=max(0.0, deadline - time.monotonic())):
+                return {}
+            while True:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    break
+                waiter.clear()
+                if not waiter.wait(timeout=min(settle, remaining)):
+                    break
+            with self._lock:
+                return dict(self._received_topics)
+        finally:
+            with self._lock:
+                self._waiters.remove(waiter)
+
+
+_persistent_clients: Dict[Tuple[str, int], PersistentBrokerClient] = {}
+_persistent_clients_lock = threading.Lock()
+
+
+def get_persistent_broker_client(host: str, port: int) -> PersistentBrokerClient:
+    key = (host, port)
+    with _persistent_clients_lock:
+        client = _persistent_clients.get(key)
+        if client is None:
+            client = PersistentBrokerClient(host, port)
+            _persistent_clients[key] = client
+        return client
+
+
 class InternalBrokerPublisher:
     def __init__(self) -> None:
         try:

packages/helpermodules/broker_test.py

@@ -0,0 +1,214 @@
+import errno
+import threading
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from helpermodules import broker
+from helpermodules.exceptions.os import handle_os_error
+from helpermodules.broker import PersistentBrokerClient, get_persistent_broker_client
+
+
+@pytest.fixture
+def mock_mqtt_client():
+    with patch("helpermodules.broker.mqtt.Client") as mock_client_cls:
+        instance = MagicMock()
+        instance.is_connected.return_value = True
+        mock_client_cls.return_value = instance
+        yield instance
+
+
+class TestPersistentBrokerClient:
+    def test_publish_does_not_raise_during_initial_connect_grace_period(self, mock_mqtt_client):
+        """connect_async()/loop_start() verbinden im Hintergrund - direkt nach dem Konstruktor ist
+        is_connected() so gut wie immer noch False, ohne dass das ein echter Fehler waere (siehe
+        Review-Kommentar zu PR #3996: der allererste publish() an einen neuen Host loeste sonst nach
+        jedem Neustart praktisch garantiert einen falschen Fehler aus)."""
+        mock_mqtt_client.is_connected.return_value = False
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+
+        client.publish("some/topic", "1")
+
+        mock_mqtt_client.publish.assert_not_called()
+
+    def test_publish_raises_once_initial_connect_grace_period_expires(self, mock_mqtt_client):
+        mock_mqtt_client.is_connected.return_value = False
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+        client._created_at -= broker.INITIAL_CONNECT_GRACE + 1
+
+        with pytest.raises(ConnectionError) as excinfo:
+            client.publish("some/topic", "1")
+
+        mock_mqtt_client.publish.assert_not_called()
+        # errno muss gesetzt sein, sonst faellt der bestehende OSError-Handler
+        # (helpermodules/exceptions/os.py) auf die nichtssagende "Unbekannter Fehler"-Meldung zurueck,
+        # statt die schon vorhandene, uebersetzte "Verbindung zum Host fehlgeschlagen"-Meldung zu nutzen.
+        assert excinfo.value.errno == errno.EHOSTUNREACH
+        assert handle_os_error(excinfo.value) == (
+            "Die Verbindung zum Host ist fehlgeschlagen. Überprüfe Adresse und Netzwerk.")
+
+    def test_publish_raises_immediately_once_connection_has_dropped(self, mock_mqtt_client):
+        """Eine Verbindung, die schon einmal stand und dann abbricht, ist ein echter Ausfall - dafuer
+        gibt es keine Kulanzfrist, das muss weiterhin sofort erkannt werden."""
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+        client._on_connect(mock_mqtt_client, None, None, 0)
+        mock_mqtt_client.is_connected.return_value = False
+
+        with pytest.raises(ConnectionError):
+            client.publish("some/topic", "1")
+
+    def test_publish_forwards_when_connected(self, mock_mqtt_client):
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+
+        client.publish("some/topic", "1", qos=0, retain=True)
+
+        mock_mqtt_client.publish.assert_called_once_with("some/topic", "1", qos=0, retain=True)
+
+    def test_ensure_subscribed_is_idempotent(self, mock_mqtt_client):
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+
+        client.ensure_subscribed("openWB/internal_chargepoint/0/get/#")
+        client.ensure_subscribed("openWB/internal_chargepoint/0/get/#")
+        client.ensure_subscribed("openWB/internal_chargepoint/1/get/#")
+
+        assert mock_mqtt_client.subscribe.call_count == 2
+
+    def test_on_connect_resubscribes_all_known_topics(self, mock_mqtt_client):
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+        client.ensure_subscribed("topic/a")
+        client.ensure_subscribed("topic/b")
+        mock_mqtt_client.subscribe.reset_mock()
+
+        client._on_connect(mock_mqtt_client, None, None, 0)
+
+        assert mock_mqtt_client.subscribe.call_count == 2
+
+    def test_wait_for_fresh_data_returns_empty_dict_when_nothing_arrives(self, mock_mqtt_client):
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+
+        result = client.wait_for_fresh_data(max_wait=0.05, settle=0.01)
+
+        assert result == {}
+
+    def test_wait_for_fresh_data_returns_accumulated_state_once_a_message_arrives(self, mock_mqtt_client):
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+
+        def deliver_message():
+            message = MagicMock()
+            message.topic = "openWB/internal_chargepoint/0/get/power"
+            message.payload = b"1500"
+            client._on_message(mock_mqtt_client, None, message)
+
+        threading.Timer(0.02, deliver_message).start()
+
+        result = client.wait_for_fresh_data(max_wait=1, settle=0.05)
+
+        assert result == {"openWB/internal_chargepoint/0/get/power": 1500}
+
+    def test_wait_for_fresh_data_keeps_previously_received_topics(self, mock_mqtt_client):
+        """Ein Zyklus mit nur teilweise neuen Daten darf frueher empfangene Topics nicht verlieren -
+        genau das war die Ursache des Bugs, den PR #3976 zu beheben versuchte."""
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+        first_message = MagicMock()
+        first_message.topic = "openWB/internal_chargepoint/0/get/power"
+        first_message.payload = b"1500"
+        client._on_message(mock_mqtt_client, None, first_message)
+
+        def deliver_second_message():
+            message = MagicMock()
+            message.topic = "openWB/internal_chargepoint/0/get/currents"
+            message.payload = b"[1.0, 1.0, 1.0]"
+            client._on_message(mock_mqtt_client, None, message)
+
+        threading.Timer(0.02, deliver_second_message).start()
+
+        result = client.wait_for_fresh_data(max_wait=1, settle=0.05)
+
+        assert result == {
+            "openWB/internal_chargepoint/0/get/power": 1500,
+            "openWB/internal_chargepoint/0/get/currents": [1.0, 1.0, 1.0],
+        }
+
+    def test_wait_for_fresh_data_registers_and_deregisters_its_own_waiter(self, mock_mqtt_client):
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+
+        client.wait_for_fresh_data(max_wait=0.02, settle=0.01)
+
+        assert client._waiters == []
+
+    def test_on_message_notifies_all_concurrent_waiters(self, mock_mqtt_client):
+        """Bei zwei Ladepunkten an einem Secondary (duo_num 0/1, gemeinsame Verbindung) laufen
+        mehrere wait_for_fresh_data()-Aufrufe gleichzeitig auf demselben PersistentBrokerClient -
+        eine eingehende Nachricht muss alle wecken, nicht nur den zuerst registrierten."""
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+        waiter_a = threading.Event()
+        waiter_b = threading.Event()
+        client._waiters.extend([waiter_a, waiter_b])
+
+        message = MagicMock()
+        message.topic = "some/topic"
+        message.payload = b"1"
+        client._on_message(mock_mqtt_client, None, message)
+
+        assert waiter_a.is_set()
+        assert waiter_b.is_set()
+
+    def test_wait_for_fresh_data_both_duo_waiters_receive_their_data(self, mock_mqtt_client):
+        """Reproduziert das im Review zu PR #3996 beschriebene Szenario: zwei Ladepunkte am selben
+        Secondary (duo_num 0/1) rufen wait_for_fresh_data() auf demselben PersistentBrokerClient
+        nahezu gleichzeitig auf. Mit einem einzigen gemeinsamen Event konnte ein Aufrufer per eigenem
+        clear() dem anderen den Wakeup wegnehmen (verpasste Daten trotz eingetroffener Nachricht) -
+        mit einem Event je Aufrufer kann das nicht mehr passieren."""
+        client = PersistentBrokerClient("10.0.0.5", 1883)
+        results = {}
+
+        def waiter(name):
+            results[name] = client.wait_for_fresh_data(max_wait=1, settle=0.05)
+
+        t1 = threading.Thread(target=waiter, args=("cp0",))
+        t2 = threading.Thread(target=waiter, args=("cp1",))
+        t1.start()
+        t2.start()
+
+        deadline = time.monotonic() + 1
+        while len(client._waiters) < 2 and time.monotonic() < deadline:
+            time.sleep(0.005)
+        assert len(client._waiters) == 2
+
+        for topic, payload in [
+            ("openWB/internal_chargepoint/0/get/power", b"1500"),
+            ("openWB/internal_chargepoint/1/get/power", b"2500"),
+        ]:
+            message = MagicMock()
+            message.topic = topic
+            message.payload = payload
+            client._on_message(mock_mqtt_client, None, message)
+
+        t1.join(timeout=2)
+        t2.join(timeout=2)
+
+        expected = {
+            "openWB/internal_chargepoint/0/get/power": 1500,
+            "openWB/internal_chargepoint/1/get/power": 2500,
+        }
+        assert results["cp0"] == expected
+        assert results["cp1"] == expected
+
+
+class TestGetPersistentBrokerClient:
+    def test_returns_same_instance_for_same_host_and_port(self, mock_mqtt_client):
+        broker._persistent_clients.clear()
+
+        first = get_persistent_broker_client("10.0.0.5", 1883)
+        second = get_persistent_broker_client("10.0.0.5", 1883)
+
+        assert first is second
+
+    def test_returns_different_instance_for_different_host(self, mock_mqtt_client):
+        broker._persistent_clients.clear()
+
+        first = get_persistent_broker_client("10.0.0.5", 1883)
+        second = get_persistent_broker_client("10.0.0.6", 1883)
+
+        assert first is not second

packages/helpermodules/pub.py

@@ -1,8 +1,7 @@
 import json
 import logging
-import paho.mqtt.publish as publish
 
-from helpermodules.broker import InternalBrokerPublisher
+from helpermodules.broker import InternalBrokerPublisher, get_persistent_broker_client
 
 
 log = logging.getLogger(__name__)
@@ -50,4 +49,5 @@ def pub_single(topic: str, payload, hostname: str = "localhost", port: int = 188
         Pub().pub(topic, payload, qos=0, no_json=no_json, retain=retain)
         return
 
-    publish.single(topic, payload if no_json else json.dumps(payload), hostname=hostname, port=port, retain=retain)
+    get_persistent_broker_client(hostname, port).publish(
+        topic, payload if no_json else json.dumps(payload), qos=0, retain=retain)